Merge lp:~pedronis/ubuntu-push/connack into lp:ubuntu-push

[Samuele Pedroni]

introduce CONNACK message from the server after CONNECT to inform the client of connection params like ping interval

introduce CONNACK message from the server after CONNECT to inform the client of connection params like ping interval

[John Lenton (chipaca)]

Connack the Barbariack!
Meaning, I approve.

review: Approve

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-01-13 15:21:11 +0000
+++ protocol/messages.go	2014-01-15 16:18:38 +0000
@@ -36,6 +36,19 @@
 	Levels map[string]int64
 }
 
+// CONNACK message
+type ConnAckMsg struct {
+	Type   string `json:"T"`
+	Params ConnAckParams
+}
+
+// ConnAckParams carries the connection parameters from the server on
+// connection acknowledment.
+type ConnAckParams struct {
+	// ping interval formatted time.Duration
+	PingInterval string
+}
+
 // PING/PONG messages
 type PingPongMsg struct {
 	Type string `json:"T"`

=== modified file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	2014-01-13 15:21:11 +0000
+++ server/acceptance/acceptance_test.go	2014-01-15 16:18:38 +0000
@@ -86,7 +86,6 @@
 	}
 	return &ClientSession{
 		ExchangeTimeout: 100 * time.Millisecond,
-		PingInterval:    500 * time.Millisecond,
 		ServerAddr:      addr,
 		CertPEMBlock:    certPEMBlock,
 		DeviceId:        deviceId,
@@ -245,7 +244,7 @@
 	c.Assert(err, IsNil)
 	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
 		// would be 3rd ping read, based on logged traffic
-		if op == "read" && ic.totalRead >= 28 {
+		if op == "read" && ic.totalRead >= 79 {
 			// exit the sess.Run() goroutine, client will close
 			runtime.Goexit()
 		}
@@ -279,7 +278,7 @@
 	c.Assert(err, IsNil)
 	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
 		// would be pong to 2nd ping, based on logged traffic
-		if op == "write" && ic.totalRead >= 28 {
+		if op == "write" && ic.totalRead >= 67 {
 			time.Sleep(200 * time.Millisecond)
 			// exit the sess.Run() goroutine, client will close
 			runtime.Goexit()

=== modified file 'server/acceptance/acceptanceclient.go'
--- server/acceptance/acceptanceclient.go	2014-01-13 17:43:34 +0000
+++ server/acceptance/acceptanceclient.go	2014-01-15 16:18:38 +0000
@@ -34,7 +34,6 @@
 	// configuration
 	DeviceId        string
 	ServerAddr      string
-	PingInterval    time.Duration
 	ExchangeTimeout time.Duration
 	CertPEMBlock    []byte
 	ReportPings     bool
@@ -89,10 +88,19 @@
 	if err != nil {
 		return err
 	}
+	var connAck protocol.ConnAckMsg
+	err = proto.ReadMessage(&connAck)
+	if err != nil {
+		return err
+	}
+	pingInterval, err := time.ParseDuration(connAck.Params.PingInterval)
+	if err != nil {
+		return err
+	}
 	events <- fmt.Sprintf("connected %v", conn.LocalAddr())
 	var recv serverMsg
 	for {
-		deadAfter := sess.PingInterval + sess.ExchangeTimeout
+		deadAfter := pingInterval + sess.ExchangeTimeout
 		conn.SetDeadline(time.Now().Add(deadAfter))
 		err = proto.ReadMessage(&recv)
 		if err != nil {

=== modified file 'server/acceptance/cmd/acceptanceclient.go'
--- server/acceptance/cmd/acceptanceclient.go	2014-01-13 17:43:34 +0000
+++ server/acceptance/cmd/acceptanceclient.go	2014-01-15 16:18:38 +0000
@@ -33,7 +33,6 @@
 
 type configuration struct {
 	// session configuration
-	PingInterval    config.ConfigTimeDuration `json:"ping_interval"`
 	ExchangeTimeout config.ConfigTimeDuration `json:"exchange_timeout"`
 	// server connection config
 	Addr        config.ConfigHostPort
@@ -61,7 +60,6 @@
 	}
 	session := &acceptance.ClientSession{
 		ExchangeTimeout: cfg.ExchangeTimeout.TimeDuration(),
-		PingInterval:    cfg.PingInterval.TimeDuration(),
 		ServerAddr:      cfg.Addr.HostPort(),
 		DeviceId:        flag.Arg(1),
 		// flags

=== modified file 'server/session/session.go'
--- server/session/session.go	2014-01-14 14:34:20 +0000
+++ server/session/session.go	2014-01-15 16:18:38 +0000
@@ -44,6 +44,14 @@
 	if connMsg.Type != "connect" {
 		return nil, &broker.ErrAbort{"expected CONNECT message"}
 	}
+	proto.SetDeadline(time.Now().Add(cfg.ExchangeTimeout()))
+	err = proto.WriteMessage(&protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{PingInterval: cfg.PingInterval().String()},
+	})
+	if err != nil {
+		return nil, err
+	}
 	return brkr.Register(&connMsg)
 }
 

=== modified file 'server/session/session_test.go'
--- server/session/session_test.go	2014-01-14 14:34:20 +0000
+++ server/session/session_test.go	2014-01-15 16:18:38 +0000
@@ -109,7 +109,8 @@
 	return tsc.exchangeTimeout
 }
 
-var cfg5msExchangeTout = &testSessionConfig{
+var cfg10msPingInterval5msExchangeTout = &testSessionConfig{
+	pingInterval:    10 * time.Millisecond,
 	exchangeTimeout: 5 * time.Millisecond,
 }
 
@@ -153,11 +154,17 @@
 	brkr := newTestBroker()
 	go func() {
 		var err error
-		sess, err = sessionStart(tp, brkr, cfg5msExchangeTout)
+		sess, err = sessionStart(tp, brkr, cfg10msPingInterval5msExchangeTout)
 		errCh <- err
 	}()
 	c.Check(takeNext(down), Equals, "deadline 5ms")
 	up <- protocol.ConnectMsg{Type: "connect", ClientVer: "1", DeviceId: "dev-1"}
+	c.Check(takeNext(down), Equals, "deadline 5ms")
+	c.Check(takeNext(down), Equals, protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{(10 * time.Millisecond).String()},
+	})
+	up <- nil // no write error
 	err := <-errCh
 	c.Check(err, IsNil)
 	c.Check(takeNext(brkr.registration), Equals, "register dev-1")
@@ -175,21 +182,37 @@
 	brkr.err = errRegister
 	go func() {
 		var err error
-		sess, err = sessionStart(tp, brkr, cfg5msExchangeTout)
+		sess, err = sessionStart(tp, brkr, cfg10msPingInterval5msExchangeTout)
 		errCh <- err
 	}()
 	up <- protocol.ConnectMsg{Type: "connect", ClientVer: "1", DeviceId: "dev-1"}
+	takeNext(down) // CONNACK
+	up <- nil      // no write error
 	err := <-errCh
 	c.Check(err, Equals, errRegister)
 }
 
-func (s *sessionSuite) TestSessionStartErrors(c *C) {
-	up := make(chan interface{}, 5)
-	down := make(chan interface{}, 5)
-	tp := &testProtocol{up, down}
-	up <- io.ErrUnexpectedEOF
-	_, err := sessionStart(tp, nil, cfg5msExchangeTout)
-	c.Check(err, Equals, io.ErrUnexpectedEOF)
+func (s *sessionSuite) TestSessionStartReadError(c *C) {
+	up := make(chan interface{}, 5)
+	down := make(chan interface{}, 5)
+	tp := &testProtocol{up, down}
+	up <- io.ErrUnexpectedEOF
+	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout)
+	c.Check(err, Equals, io.ErrUnexpectedEOF)
+}
+
+func (s *sessionSuite) TestSessionStartWriteError(c *C) {
+	up := make(chan interface{}, 5)
+	down := make(chan interface{}, 5)
+	tp := &testProtocol{up, down}
+	up <- protocol.ConnectMsg{Type: "connect"}
+	up <- io.ErrUnexpectedEOF
+	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout)
+	c.Check(err, Equals, io.ErrUnexpectedEOF)
+	// sanity
+	c.Check(takeNext(down), Matches, "deadline.*")
+	c.Check(takeNext(down), Matches, "deadline.*")
+	c.Check(takeNext(down), FitsTypeOf, protocol.ConnAckMsg{})
 }
 
 func (s *sessionSuite) TestSessionStartMismatch(c *C) {
@@ -197,7 +220,7 @@
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	up <- protocol.ConnectMsg{Type: "what"}
-	_, err := sessionStart(tp, nil, cfg5msExchangeTout)
+	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout)
 	c.Check(err, DeepEquals, &broker.ErrAbort{"expected CONNECT message"})
 }
 
@@ -479,15 +502,24 @@
 	}()
 	io.WriteString(cli, "\x00")
 	io.WriteString(cli, "\x00\x20{\"T\":\"connect\",\"DeviceId\":\"DEV\"}")
+	// connack
 	downStream := bufio.NewReader(cli)
 	msg, err := downStream.ReadBytes(byte('}'))
 	c.Check(err, IsNil)
+	c.Check(msg, DeepEquals, []byte("\x00\x2f{\"T\":\"connack\",\"Params\":{\"PingInterval\":\"5ms\"}"))
+	// eat the last }
+	rbr, err := downStream.ReadByte()
+	c.Check(err, IsNil)
+	c.Check(rbr, Equals, byte('}'))
+	// first ping
+	msg, err = downStream.ReadBytes(byte('}'))
+	c.Check(err, IsNil)
 	c.Check(msg, DeepEquals, []byte("\x00\x0c{\"T\":\"ping\"}"))
 	c.Check(takeNext(brkr.registration), Equals, "register DEV")
 	c.Check(len(brkr.registration), Equals, 0) // not yet unregistered
 	cli.Close()
 	err = <-errCh
-	c.Check(remSrv.deadlineKind, DeepEquals, []string{"read", "both", "both"})
+	c.Check(remSrv.deadlineKind, DeepEquals, []string{"read", "both", "both", "both"})
 	c.Check(err, Equals, io.EOF)
 	c.Check(takeNext(brkr.registration), Equals, "unregister DEV")
 	// tracking