Merge lp:~chipaca/ubuntu-push/introduction-to-quantum-client-sessions into lp:ubuntu-push

[John Lenton]

The running man.

Client session, volume 4: A gentle introduction to quantum client sessions.

[lp:~chipaca/ubuntu-push/introduction-to-quantum-client-sessions]

40. By John Lenton on 2014-01-31

Merged complex-client-session into introduction-to-quantum-client-sessions.

41. By John Lenton on 2014-01-31

Merged complex-client-session into introduction-to-quantum-client-sessions.

42. By John Lenton on 2014-01-31

merged and resolved

43. By John Lenton on 2014-01-31

Merged complex-client-session into introduction-to-quantum-client-sessions.

44. By John Lenton on 2014-01-31

the good ol' proto.SetDeadline

45. By John Lenton on 2014-01-31

Merged complex-client-session into introduction-to-quantum-client-sessions.

46. By John Lenton on 2014-01-31

AckMsg still need to specify Type

47. By John Lenton on 2014-01-31

Merged complex-client-session into introduction-to-quantum-client-sessions.

48. By John Lenton on 2014-01-31

Too many deadlines! I know the feeling.

49. By John Lenton on 2014-01-31

Merged complex-client-session into introduction-to-quantum-client-sessions.

[Samuele Pedroni (pedronis)]

review: Approve

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-01-31 16:05:03 +0000
+++ client/session/session.go	2014-01-31 17:43:04 +0000
@@ -116,7 +116,6 @@
 
 // handle "ping" messages
 func (sess *ClientSession) handlePing() error {
-	sess.proto.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
 	err := sess.proto.WriteMessage(protocol.PingPongMsg{Type: "pong"})
 	if err == nil {
 		sess.Log.Debugf("ping.")
@@ -128,7 +127,6 @@
 
 // handle "broadcast" messages
 func (sess *ClientSession) handleBroadcast(bcast *serverMsg) error {
-	sess.proto.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
 	err := sess.proto.WriteMessage(protocol.AckMsg{"ack"})
 	if err != nil {
 		return err
@@ -144,3 +142,26 @@
 	}
 	return nil
 }
+
+// Run the session with the server, emits a stream of events.
+func (sess *ClientSession) run() error {
+	var err error
+	var recv serverMsg
+	for {
+		deadAfter := sess.pingInterval + sess.ExchangeTimeout
+		sess.proto.SetDeadline(time.Now().Add(deadAfter))
+		err = sess.proto.ReadMessage(&recv)
+		if err != nil {
+			return err
+		}
+		switch recv.Type {
+		case "ping":
+			err = sess.handlePing()
+		case "broadcast":
+			err = sess.handleBroadcast(&recv)
+		}
+		if err != nil {
+			return err
+		}
+	}
+}
 
=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-01-31 16:05:03 +0000
+++ client/session/session_test.go	2014-01-31 17:43:04 +0000
@@ -104,7 +104,19 @@
 }
 
 func (c *testProtocol) ReadMessage(dest interface{}) error {
-	panic("ReadMessage not implemented.")
+	switch v := takeNext(c.up).(type) {
+	case error:
+		return v
+	default:
+		// make sure JSON.Unmarshal works with dest
+		var marshalledMsg []byte
+		marshalledMsg, err := json.Marshal(v)
+		if err != nil {
+			return fmt.Errorf("can't jsonify test value %v: %s", v, err)
+		}
+		return json.Unmarshal(marshalledMsg, dest)
+	}
+	return nil
 }
 
 func (c *testProtocol) WriteMessage(src interface{}) error {
@@ -277,8 +289,7 @@
 func (s *msgSuite) TestHandlePingWorks(c *C) {
 	s.upCh <- nil // no error
 	c.Check(s.sess.handlePing(), IsNil)
-	c.Assert(len(s.downCh), Equals, 2)
-	c.Check(<-s.downCh, Equals, "deadline 1ms")
+	c.Assert(len(s.downCh), Equals, 1)
 	c.Check(<-s.downCh, Equals, protocol.PingPongMsg{Type: "pong"})
 }
 
@@ -287,8 +298,7 @@
 	s.upCh <- failure
 
 	c.Check(s.sess.handlePing(), Equals, failure)
-	c.Assert(len(s.downCh), Equals, 2)
-	c.Check(<-s.downCh, Equals, "deadline 1ms")
+	c.Assert(len(s.downCh), Equals, 1)
 	c.Check(<-s.downCh, Equals, protocol.PingPongMsg{Type: "pong"})
 }
 
@@ -306,7 +316,6 @@
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
 		}, protocol.NotificationsMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
-	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
 	c.Check(<-s.errCh, Equals, nil)
@@ -326,7 +335,6 @@
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
 		}, protocol.NotificationsMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
-	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	failure := errors.New("ACK ACK ACK")
 	s.upCh <- failure
@@ -343,9 +351,68 @@
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
 		}, protocol.NotificationsMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
-	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
 	c.Check(<-s.errCh, IsNil)
 	c.Check(len(s.sess.MsgCh), Equals, 0)
 }
+
+/****************************************************************
+  run() tests
+****************************************************************/
+
+type runSuite msgSuite
+
+var _ = Suite(&runSuite{})
+
+func (s *runSuite) SetUpTest(c *C) {
+	(*msgSuite)(s).SetUpTest(c)
+	s.sess.Connection.(*testConn).Name = "TestRun* (small r)"
+	go func() {
+		s.errCh <- s.sess.run()
+	}()
+}
+
+func (s *runSuite) TestRunReadError(c *C) {
+	s.upCh <- errors.New("Read")
+	err := <-s.errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Equals, "Read")
+}
+
+func (s *runSuite) TestRunPing(c *C) {
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	s.upCh <- protocol.PingPongMsg{Type: "ping"}
+	c.Check(takeNext(s.downCh), Equals, protocol.PingPongMsg{Type: "pong"})
+	failure := errors.New("pong")
+	s.upCh <- failure
+	c.Check(<-s.errCh, Equals, failure)
+}
+
+func (s *runSuite) TestRunLoopsDaLoop(c *C) {
+	for i := 1; i < 10; i++ {
+		c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+		s.upCh <- protocol.PingPongMsg{Type: "ping"}
+		c.Check(takeNext(s.downCh), Equals, protocol.PingPongMsg{Type: "pong"})
+		s.upCh <- nil
+	}
+	failure := errors.New("pong")
+	s.upCh <- failure
+	c.Check(<-s.errCh, Equals, failure)
+}
+
+func (s *runSuite) TestRunBroadcast(c *C) {
+	b := &protocol.BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "--ignored--",
+		ChanId:   "0",
+		TopLevel: 2,
+		Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
+	}
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	s.upCh <- b
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	failure := errors.New("ack")
+	s.upCh <- failure
+	c.Check(<-s.errCh, Equals, failure)
+}