Merge lp:~pedronis/ubuntu-push/unicast-in-session into lp:ubuntu-push/automatic

[Samuele Pedroni]

first pass of unicast support in session

first pass of unicast support in session

(persistence for recent message ids will come in the next branch)

[John Lenton (chipaca)]

Lovely.

review: Approve

=== modified file 'client/client.go'
--- client/client.go	2014-05-14 12:28:56 +0000
+++ client/client.go	2014-05-14 13:37:29 +0000
@@ -235,7 +235,7 @@
 	}
 }
 
-// filterNotification finds out if the notification is about an actual
+// filterBroadcastNotification finds out if the notification is about an actual
 // upgrade for the device. It expects msg.Decoded entries to look
 // like:
 //
@@ -243,7 +243,7 @@
 // "IMAGE-CHANNEL/DEVICE-MODEL": [BUILD-NUMBER, CHANNEL-ALIAS]
 // ...
 // }
-func (client *PushClient) filterNotification(msg *session.Notification) bool {
+func (client *PushClient) filterBroadcastNotification(msg *session.BroadcastNotification) bool {
 	n := len(msg.Decoded)
 	if n == 0 {
 		return false
@@ -279,8 +279,8 @@
 }
 
 // handleNotification deals with receiving a notification
-func (client *PushClient) handleNotification(msg *session.Notification) error {
-	if !client.filterNotification(msg) {
+func (client *PushClient) handleBroadcastNotification(msg *session.BroadcastNotification) error {
+	if !client.filterBroadcastNotification(msg) {
 		return nil
 	}
 	action_id := ACTION_ID_SNOWFLAKE
@@ -317,15 +317,17 @@
 }
 
 // doLoop connects events with their handlers
-func (client *PushClient) doLoop(connhandler func(bool), clickhandler func(string) error, notifhandler func(*session.Notification) error, errhandler func(error)) {
+func (client *PushClient) doLoop(connhandler func(bool), clickhandler func(string) error, bcasthandler func(*session.BroadcastNotification) error, errhandler func(error)) {
 	for {
 		select {
 		case state := <-client.connCh:
 			connhandler(state)
 		case action := <-client.actionsCh:
 			clickhandler(action.ActionId)
-		case msg := <-client.session.MsgCh:
-			notifhandler(msg)
+		case bcast := <-client.session.BroadcastCh:
+			bcasthandler(bcast)
+		case _ = <-client.session.NotificationsCh:
+			// xxx implement me
 		case err := <-client.session.ErrCh:
 			errhandler(err)
 		case count := <-client.sessionConnectedCh:
@@ -348,7 +350,7 @@
 // Loop calls doLoop with the "real" handlers
 func (client *PushClient) Loop() {
 	client.doLoop(client.handleConnState, client.handleClick,
-		client.handleNotification, client.handleErr)
+		client.handleBroadcastNotification, client.handleErr)
 }
 
 func (client *PushClient) startService() error {

=== modified file 'client/client_test.go'
--- client/client_test.go	2014-05-14 12:28:56 +0000
+++ client/client_test.go	2014-05-14 13:37:29 +0000
@@ -491,7 +491,7 @@
 }
 
 /*****************************************************************
-   filterNotification tests
+   filterBroadcastNotification tests
 ******************************************************************/
 
 var siInfoRes = &systemimage.InfoResult{
@@ -501,23 +501,23 @@
 	LastUpdate:  "Unknown",
 }
 
-func (cs *clientSuite) TestFilterNotification(c *C) {
+func (cs *clientSuite) TestFilterBroadcastNotification(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	// empty
-	msg := &session.Notification{}
-	c.Check(cli.filterNotification(msg), Equals, false)
+	msg := &session.BroadcastNotification{}
+	c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	// same build number
-	msg = &session.Notification{
+	msg = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
 			},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, false)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	// higher build number and pick last
-	msg = &session.Notification{
+	msg = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
@@ -527,9 +527,9 @@
 			},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, true)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, true)
 	// going backward by a margin, assume switch of alias
-	msg = &session.Notification{
+	msg = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
@@ -539,47 +539,47 @@
 			},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, true)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, true)
 }
 
-func (cs *clientSuite) TestFilterNotificationRobust(c *C) {
+func (cs *clientSuite) TestFilterBroadcastNotificationRobust(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
-	msg := &session.Notification{
+	msg := &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, false)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	for _, broken := range []interface{}{
 		5,
 		[]interface{}{},
 		[]interface{}{55},
 	} {
-		msg := &session.Notification{
+		msg := &session.BroadcastNotification{
 			Decoded: []map[string]interface{}{
 				map[string]interface{}{
 					"daily/mako": broken,
 				},
 			},
 		}
-		c.Check(cli.filterNotification(msg), Equals, false)
+		c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	}
 }
 
 /*****************************************************************
-    handleNotification tests
+    handleBroadcastNotification tests
 ******************************************************************/
 
 var (
-	positiveNotification = &session.Notification{
+	positiveBroadcastNotification = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(103), "tubular"},
 			},
 		},
 	}
-	negativeNotification = &session.Notification{
+	negativeBroadcastNotification = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
@@ -588,13 +588,13 @@
 	}
 )
 
-func (cs *clientSuite) TestHandleNotification(c *C) {
+func (cs *clientSuite) TestHandleBroadcastNotification(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(true), uint32(1))
 	cli.notificationsEndp = endp
 	cli.log = cs.log
-	c.Check(cli.handleNotification(positiveNotification), IsNil)
+	c.Check(cli.handleBroadcastNotification(positiveBroadcastNotification), IsNil)
 	// check we sent the notification
 	args := testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 1)
@@ -602,26 +602,26 @@
 	c.Check(cs.log.Captured(), Matches, `.* got notification id \d+\s*`)
 }
 
-func (cs *clientSuite) TestHandleNotificationNothingToDo(c *C) {
+func (cs *clientSuite) TestHandleBroadcastNotificationNothingToDo(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(true), uint32(1))
 	cli.notificationsEndp = endp
 	cli.log = cs.log
-	c.Check(cli.handleNotification(negativeNotification), IsNil)
+	c.Check(cli.handleBroadcastNotification(negativeBroadcastNotification), IsNil)
 	// check we sent the notification
 	args := testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 0)
 	c.Check(cs.log.Captured(), Matches, "")
 }
 
-func (cs *clientSuite) TestHandleNotificationFail(c *C) {
+func (cs *clientSuite) TestHandleBroadcastNotificationFail(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	cli.log = cs.log
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(false))
 	cli.notificationsEndp = endp
-	c.Check(cli.handleNotification(positiveNotification), NotNil)
+	c.Check(cli.handleBroadcastNotification(positiveBroadcastNotification), NotNil)
 }
 
 /*****************************************************************
@@ -660,7 +660,7 @@
 	c.Assert(cli.initSession(), IsNil)
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) { ch <- true }, func(_ string) error { return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
+	go cli.doLoop(func(bool) { ch <- true }, func(_ string) error { return nil }, func(_ *session.BroadcastNotification) error { return nil }, func(error) {})
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -674,7 +674,7 @@
 	cli.actionsCh = aCh
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func(_ string) error { ch <- true; return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
+	go cli.doLoop(func(bool) {}, func(_ string) error { ch <- true; return nil }, func(_ *session.BroadcastNotification) error { return nil }, func(error) {})
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -683,11 +683,11 @@
 	cli.log = cs.log
 	cli.systemImageInfo = siInfoRes
 	c.Assert(cli.initSession(), IsNil)
-	cli.session.MsgCh = make(chan *session.Notification, 1)
-	cli.session.MsgCh <- &session.Notification{}
+	cli.session.BroadcastCh = make(chan *session.BroadcastNotification, 1)
+	cli.session.BroadcastCh <- &session.BroadcastNotification{}
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.Notification) error { ch <- true; return nil }, func(error) {})
+	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.BroadcastNotification) error { ch <- true; return nil }, func(error) {})
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -700,7 +700,7 @@
 	cli.session.ErrCh <- nil
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.Notification) error { return nil }, func(error) { ch <- true })
+	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.BroadcastNotification) error { return nil }, func(error) { ch <- true })
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -750,7 +750,7 @@
 	cli.systemImageInfo = siInfoRes
 	c.Assert(cli.initSession(), IsNil)
 
-	cli.session.MsgCh = make(chan *session.Notification)
+	cli.session.BroadcastCh = make(chan *session.BroadcastNotification)
 	cli.session.ErrCh = make(chan error)
 
 	// we use tick() to make sure things have been through the
@@ -783,8 +783,8 @@
 	tick()
 	c.Check(cli.hasConnectivity, Equals, false)
 
-	//  * session.MsgCh to the notifications handler
-	cli.session.MsgCh <- positiveNotification
+	//  * session.BroadcastCh to the notifications handler
+	cli.session.BroadcastCh <- positiveBroadcastNotification
 	tick()
 	nargs := testibus.GetCallArgs(cli.notificationsEndp)
 	c.Check(nargs, HasLen, 1)

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-05-02 10:42:16 +0000
+++ client/session/session.go	2014-05-14 13:37:29 +0000
@@ -43,7 +43,7 @@
 	wireVersionBytes = []byte{protocol.ProtocolWireVersion}
 )
 
-type Notification struct {
+type BroadcastNotification struct {
 	TopLevel int64
 	Decoded  []map[string]interface{}
 }
@@ -116,9 +116,10 @@
 	pingInterval time.Duration
 	retrier      util.AutoRedialer
 	// status
-	stateP *uint32
-	ErrCh  chan error
-	MsgCh  chan *Notification
+	stateP          *uint32
+	ErrCh           chan error
+	BroadcastCh     chan *BroadcastNotification
+	NotificationsCh chan *protocol.Notification
 	// authorization
 	auth string
 	// autoredial knobs
@@ -370,7 +371,7 @@
 	return err
 }
 
-func (sess *ClientSession) decodeBroadcast(bcast *serverMsg) *Notification {
+func (sess *ClientSession) decodeBroadcast(bcast *serverMsg) *BroadcastNotification {
 	decoded := make([]map[string]interface{}, 0)
 	for _, p := range bcast.Payloads {
 		var v map[string]interface{}
@@ -381,7 +382,7 @@
 		}
 		decoded = append(decoded, v)
 	}
-	return &Notification{
+	return &BroadcastNotification{
 		TopLevel: bcast.TopLevel,
 		Decoded:  decoded,
 	}
@@ -409,15 +410,37 @@
 		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.Log.Debugf("sending it over")
-		sess.MsgCh <- sess.decodeBroadcast(bcast)
-		sess.Log.Debugf("sent it over")
+		sess.Log.Debugf("sending bcast over")
+		sess.BroadcastCh <- sess.decodeBroadcast(bcast)
+		sess.Log.Debugf("sent bcast over")
 	} else {
 		sess.Log.Debugf("what is this weird channel, %#v?", bcast.ChanId)
 	}
 	return nil
 }
 
+// handle "notifications" messages
+func (sess *ClientSession) handleNotifications(ucast *serverMsg) error {
+	// the server assumes if we ack the broadcast, we've updated
+	// our state. Hence the order.
+	err := sess.proto.WriteMessage(protocol.AckMsg{"ack"})
+	if err != nil {
+		sess.setState(Error)
+		sess.Log.Errorf("unable to ack notifications: %s", err)
+		return err
+	}
+	sess.clearShouldDelay()
+	for i := range ucast.Notifications {
+		notif := &ucast.Notifications[i]
+		sess.Log.Debugf("unicast app:%v msg:%s payload:%s",
+			notif.AppId, notif.MsgId, notif.Payload)
+		sess.Log.Debugf("sending ucast over")
+		sess.NotificationsCh <- notif
+		sess.Log.Debugf("sent ucast over")
+	}
+	return nil
+}
+
 // handle "connbroken" messages
 func (sess *ClientSession) handleConnBroken(connBroken *serverMsg) error {
 	sess.setState(Error)
@@ -449,6 +472,8 @@
 			err = sess.handlePing()
 		case "broadcast":
 			err = sess.handleBroadcast(&recv)
+		case "notifications":
+			err = sess.handleNotifications(&recv)
 		case "connbroken":
 			err = sess.handleConnBroken(&recv)
 		}
@@ -534,7 +559,8 @@
 		err = starter()
 		if err == nil {
 			sess.ErrCh = make(chan error, 1)
-			sess.MsgCh = make(chan *Notification)
+			sess.BroadcastCh = make(chan *BroadcastNotification)
+			sess.NotificationsCh = make(chan *protocol.Notification)
 			go func() { sess.ErrCh <- looper() }()
 		}
 	}

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-05-02 10:42:16 +0000
+++ client/session/session_test.go	2014-05-14 13:37:29 +0000
@@ -645,7 +645,8 @@
 	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)
+	s.sess.BroadcastCh = make(chan *BroadcastNotification, 5)
+	s.sess.NotificationsCh = make(chan *protocol.Notification, 5)
 }
 
 func (s *msgSuite) TestHandlePingWorks(c *C) {
@@ -704,8 +705,8 @@
 	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, DeepEquals, &Notification{
+	c.Assert(len(s.sess.BroadcastCh), Equals, 1)
+	c.Check(<-s.sess.BroadcastCh, DeepEquals, &BroadcastNotification{
 		TopLevel: 2,
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
@@ -752,7 +753,7 @@
 	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)
+	c.Check(len(s.sess.BroadcastCh), Equals, 0)
 }
 
 func (s *msgSuite) TestHandleBroadcastWrongBrokenLevelmap(c *C) {
@@ -770,7 +771,7 @@
 	// start returns with error
 	c.Check(<-s.errCh, Not(Equals), nil)
 	// no message sent out
-	c.Check(len(s.sess.MsgCh), Equals, 0)
+	c.Check(len(s.sess.BroadcastCh), Equals, 0)
 	// and nak'ed it
 	c.Check(len(s.downCh), Equals, 1)
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"nak"})
@@ -803,6 +804,59 @@
 }
 
 /****************************************************************
+  handleNotifications() tests
+****************************************************************/
+
+func (s *msgSuite) TestHandleNotificationsWorks(c *C) {
+	s.sess.setShouldDelay()
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	n2 := protocol.Notification{
+		AppId:   "app2",
+		MsgId:   "b",
+		Payload: json.RawMessage(`{"m": 2}`),
+	}
+	msg := serverMsg{"notifications",
+		protocol.BroadcastMsg{},
+		protocol.NotificationsMsg{
+			Notifications: []protocol.Notification{n1, n2},
+		}, protocol.ConnBrokenMsg{}}
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	s.upCh <- nil // ack ok
+	c.Check(<-s.errCh, Equals, nil)
+	c.Check(s.sess.ShouldDelay(), Equals, false)
+	c.Assert(len(s.sess.NotificationsCh), Equals, 2)
+	c.Check(<-s.sess.NotificationsCh, DeepEquals, &n1)
+	c.Check(<-s.sess.NotificationsCh, DeepEquals, &n2)
+}
+
+func (s *msgSuite) TestHandleNotificationsBadAckWrite(c *C) {
+	s.sess.setShouldDelay()
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	msg := serverMsg{"notifications",
+		protocol.BroadcastMsg{},
+		protocol.NotificationsMsg{
+			Notifications: []protocol.Notification{n1},
+		}, protocol.ConnBrokenMsg{}}
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	failure := errors.New("ACK ACK ACK")
+	s.upCh <- failure
+	c.Assert(<-s.errCh, Equals, failure)
+	c.Check(s.sess.State(), Equals, Error)
+	// didn't get to clear
+	c.Check(s.sess.ShouldDelay(), Equals, true)
+}
+
+/****************************************************************
   handleConnBroken() tests
 ****************************************************************/
 
@@ -897,6 +951,26 @@
 	c.Check(<-s.errCh, Equals, failure)
 }
 
+func (s *loopSuite) TestLoopNotifications(c *C) {
+	c.Check(s.sess.State(), Equals, Running)
+
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	msg := &protocol.NotificationsMsg{
+		Type:          "notifications",
+		Notifications: []protocol.Notification{n1},
+	}
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	s.upCh <- msg
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	failure := errors.New("ack")
+	s.upCh <- failure
+	c.Check(<-s.errCh, Equals, failure)
+}
+
 func (s *loopSuite) TestLoopConnBroken(c *C) {
 	c.Check(s.sess.State(), Equals, Running)
 	broken := protocol.ConnBrokenMsg{
@@ -1155,24 +1229,24 @@
 func (cs *clientSessionSuite) TestRunRunsEvenIfLoopFails(c *C) {
 	sess, err := NewSession("", dummyConf, "wah", cs.lvls, cs.log)
 	c.Assert(err, IsNil)
-	// just to make a point: until here we haven't set ErrCh & MsgCh (no
+	// just to make a point: until here we haven't set ErrCh & BroadcastCh (no
 	// biggie if this stops being true)
 	c.Check(sess.ErrCh, IsNil)
-	c.Check(sess.MsgCh, IsNil)
+	c.Check(sess.BroadcastCh, IsNil)
 	failureCh := make(chan error) // must be unbuffered
-	notf := &Notification{}
+	notf := &BroadcastNotification{}
 	err = sess.run(
 		func() {},
 		func() error { return nil },
 		func() error { return nil },
 		func() error { return nil },
 		func() error { return nil },
-		func() error { sess.MsgCh <- notf; return <-failureCh })
+		func() error { sess.BroadcastCh <- notf; return <-failureCh })
 	c.Check(err, Equals, nil)
 	// if run doesn't error it sets up the channels
 	c.Assert(sess.ErrCh, NotNil)
-	c.Assert(sess.MsgCh, NotNil)
-	c.Check(<-sess.MsgCh, Equals, notf)
+	c.Assert(sess.BroadcastCh, NotNil)
+	c.Check(<-sess.BroadcastCh, Equals, notf)
 	failure := errors.New("TestRunRunsEvenIfLoopFails")
 	failureCh <- failure
 	c.Check(<-sess.ErrCh, Equals, failure)
@@ -1375,7 +1449,7 @@
 	c.Check(takeNext(downCh), Equals, protocol.AckMsg{"ack"})
 	upCh <- nil
 	// ...get bubbled up,
-	c.Check(<-sess.MsgCh, NotNil)
+	c.Check(<-sess.BroadcastCh, NotNil)
 	// and their TopLevel remembered
 	levels, err := sess.Levels.GetAll()
 	c.Check(err, IsNil)