Merge lp:~chipaca/ubuntu-push/complex-client-session into lp:ubuntu-push

[John Lenton]

Broadcasting your horizons.

Client session, volume 3: Client session in the complex plane.

[Samuele Pedroni (pedronis)]

immaterial but for clarity, there's a AckMsg in protocol now:

33 + err := sess.proto.WriteMessage(protocol.PingPongMsg{Type: "ack"})

review: Approve

[John Lenton (chipaca)]

review: Approve

[Ubuntu One Auto Pilot (otto-pilot)]

The attempt to merge lp:~chipaca/ubuntu-push/complex-client-session into lp:ubuntu-push failed. Below is the output from the failed tests.

mkdir -p /mnt/tarmac/cache/ubuntu-push/go-ws/bin
mkdir -p /mnt/tarmac/cache/ubuntu-push/go-ws/pkg
go get -u launchpad.net/godeps
go get -d -u launchpad.net/gocheck launchpad.net/go-dbus/v1
/mnt/tarmac/cache/ubuntu-push/go-ws/bin/godeps -u dependencies.tsv
"/mnt/tarmac/cache/ubuntu-push/go-ws/src/launchpad.net/gocheck" now at <email address hidden>
go install launchpad.net/gocheck launchpad.net/go-dbus/v1
go test launchpad.net/ubuntu-push/...
ok launchpad.net/ubuntu-push/bus 0.027s
ok launchpad.net/ubuntu-push/bus/connectivity 0.086s
ok launchpad.net/ubuntu-push/bus/networkmanager 0.065s
ok launchpad.net/ubuntu-push/bus/notifications 0.020s
ok launchpad.net/ubuntu-push/bus/testing 0.018s
ok launchpad.net/ubuntu-push/bus/urldispatcher 0.066s
ok launchpad.net/ubuntu-push/client/session 0.040s
ok launchpad.net/ubuntu-push/client/session/levelmap 0.005s
ok launchpad.net/ubuntu-push/config 0.009s
ok launchpad.net/ubuntu-push/logger 0.008s
ok launchpad.net/ubuntu-push/protocol 0.011s
ok launchpad.net/ubuntu-push/server 0.026s
ok launchpad.net/ubuntu-push/server/acceptance 0.008s
? launchpad.net/ubuntu-push/server/acceptance/cmd [no test files]
ok launchpad.net/ubuntu-push/server/api 0.035s
ok launchpad.net/ubuntu-push/server/broker 0.017s
ok launchpad.net/ubuntu-push/server/broker/simple 0.016s
? launchpad.net/ubuntu-push/server/broker/testing [no test files]
? launchpad.net/ubuntu-push/server/broker/testsuite [no test files]
? launchpad.net/ubuntu-push/server/dev [no test files]
ok launchpad.net/ubuntu-push/server/listener 0.497s

----------------------------------------------------------------------
FAIL: session_test.go:480: sessionSuite.TestSessionWire

session_test.go:511:
    c.Check(err, Equals, io.EOF)
... obtained *net.OpError = &net.OpError{Op:"read", Net:"tcp", Addr:(*net.TCPAddr)(0xc2100378a0), Err:(*net.timeoutError)(0x88f718)} ("read tcp 127.0.0.1:57857: i/o timeout")
... expected *errors.errorString = &errors.errorString{s:"EOF"} ("EOF")

session_test.go:514:
    // tracking
    c.Check(buf.String(), Matches, `.*connected.*\n.*registered DEV.*\n.*ended with: EOF\n`)
... value string = "" +
... "2014/01/31 14:57:34.417553 DEBUG session(7970372a523ef3) connected 127.0.0.1:57857\n" +
... "2014/01/31 14:57:34.417849 INFO session(7970372a523ef3) registered DEV\n" +
... "2014/01/31 14:57:34.430397 DEBUG session(7970372a523ef3) ended with: read tcp 127.0.0.1:57857: i/o timeout\n"
... regex string = ".*connected.*\\n.*registered DEV.*\\n.*ended with: EOF\\n"

OOPS: 22 passed, 1 FAILED
--- FAIL: TestSession (0.10 seconds)
FAIL
FAIL launchpad.net/ubuntu-push/server/session 0.131s
ok launchpad.net/ubuntu-push/server/store 0.019s
? launchpad.net/ubuntu-push/testing [no test files]
ok launchpad.net/ubuntu-push/testing/condition 0.009s
ok launchpad.net/ubuntu-push/util 0.052s
ok launchpad.net/ubuntu-push/whoopsie/identifier 0.018s
ok launchpad.net/ubuntu-push/whoopsie/identifier/testing 0.007s

make: *** [check...

[Samuele Pedroni (pedronis)]

review: Needs Fixing

[Ubuntu One Auto Pilot (otto-pilot)]

Voting does not meet specified criteria. Required: Approve >= 1, Disapprove == 0, Needs Fixing == 0, Needs Information == 0, Resubmit == 0, Pending == 0. Got: 1 Approve, 1 Needs Fixing.

[Samuele Pedroni (pedronis)]

AckMsg still needs to specify a type

[Samuele Pedroni (pedronis)]

review: Approve

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-01-31 14:06:31 +0000
+++ client/session/session.go	2014-01-31 16:06:58 +0000
@@ -35,6 +35,12 @@
 	// something something something
 }
 
+type serverMsg struct {
+	Type string `json:"T"`
+	protocol.BroadcastMsg
+	protocol.NotificationsMsg
+}
+
 // ClienSession holds a client<->server session and its configuration.
 type ClientSession struct {
 	// configuration
@@ -112,6 +118,29 @@
 func (sess *ClientSession) handlePing() error {
 	sess.proto.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
 	err := sess.proto.WriteMessage(protocol.PingPongMsg{Type: "pong"})
-	sess.Log.Debugf("Ping.")
+	if err == nil {
+		sess.Log.Debugf("ping.")
+	} else {
+		sess.Log.Errorf("unable to pong: %s", err)
+	}
 	return err
 }
+
+// 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
+	}
+	sess.Log.Debugf("broadcast chan:%v app:%v topLevel:%d payloads:%s",
+		bcast.ChanId, bcast.AppId, bcast.TopLevel, bcast.Payloads)
+	if bcast.ChanId == protocol.SystemChannelId {
+		// the system channel id, the only one we care about for now
+		sess.Levels.Set(bcast.ChanId, bcast.TopLevel)
+		sess.MsgCh <- &Notification{}
+	} else {
+		sess.Log.Debugf("what is this weird channel, %s?", bcast.ChanId)
+	}
+	return nil
+}

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-01-31 14:15:47 +0000
+++ client/session/session_test.go	2014-01-31 16:06:58 +0000
@@ -263,18 +263,19 @@
 
 func (s *msgSuite) SetUpTest(c *C) {
 	var err error
-	s.sess, err = NewSession("", nil, 0, "wah", debuglog)
+	s.sess, err = NewSession("", nil, time.Millisecond, "wah", debuglog)
 	c.Assert(err, IsNil)
-	s.sess.Connection = &testConn{Name: "TestRun* (small r)"}
+	s.sess.Connection = &testConn{Name: "TestHandle*"}
 	s.errCh = make(chan error, 1)
 	s.upCh = make(chan interface{}, 5)
 	s.downCh = make(chan interface{}, 5)
 	s.sess.proto = &testProtocol{up: s.upCh, down: s.downCh}
+	// make the message channel buffered
+	s.sess.MsgCh = make(chan *Notification, 5)
 }
 
 func (s *msgSuite) TestHandlePingWorks(c *C) {
 	s.upCh <- nil // no error
-	s.sess.ExchangeTimeout = time.Millisecond
 	c.Check(s.sess.handlePing(), IsNil)
 	c.Assert(len(s.downCh), Equals, 2)
 	c.Check(<-s.downCh, Equals, "deadline 1ms")
@@ -287,6 +288,64 @@
 
 	c.Check(s.sess.handlePing(), Equals, failure)
 	c.Assert(len(s.downCh), Equals, 2)
-	c.Check(<-s.downCh, Equals, "deadline 0")
+	c.Check(<-s.downCh, Equals, "deadline 1ms")
 	c.Check(<-s.downCh, Equals, protocol.PingPongMsg{Type: "pong"})
 }
+
+/****************************************************************
+  handleBroadcast() tests
+****************************************************************/
+
+func (s *msgSuite) TestHandleBroadcastWorks(c *C) {
+	msg := serverMsg{"broadcast",
+		protocol.BroadcastMsg{
+			Type:     "broadcast",
+			AppId:    "--ignored--",
+			ChanId:   "0",
+			TopLevel: 2,
+			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)
+	c.Assert(len(s.sess.MsgCh), Equals, 1)
+	c.Check(<-s.sess.MsgCh, Equals, &Notification{})
+	// and finally, the session keeps track of the levels
+	c.Check(s.sess.Levels.GetAll(), DeepEquals, map[string]int64{"0": 2})
+}
+
+func (s *msgSuite) TestHandleBroadcastBadAckWrite(c *C) {
+	msg := serverMsg{"broadcast",
+		protocol.BroadcastMsg{
+			Type:     "broadcast",
+			AppId:    "APP",
+			ChanId:   "0",
+			TopLevel: 2,
+			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
+	c.Assert(<-s.errCh, Equals, failure)
+}
+
+func (s *msgSuite) TestHandleBroadcastWrongChannel(c *C) {
+	msg := serverMsg{"broadcast",
+		protocol.BroadcastMsg{
+			Type:     "broadcast",
+			AppId:    "APP",
+			ChanId:   "something awful",
+			TopLevel: 2,
+			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)
+}