Merge lp:~chipaca/ubuntu-push/prepersistance into lp:ubuntu-push

[John Lenton]

change the levelmap interface to return errors

In preparation for the persistence branch, change the levelmap interface to (also) return errors.

[lp:~chipaca/ubuntu-push/prepersistance]

61. By John Lenton on 2014-02-08

Merged client-constructor into prepersistance.

[Samuele Pedroni (pedronis)]

109 - err := sess.proto.WriteMessage(protocol.AckMsg{"ack"})
110 + err := sess.Levels.Set(bcast.ChanId, bcast.TopLevel)
111 + if err != nil {
112 + sess.setState(Error)
113 + sess.Log.Errorf("unable to set level: %v", err)
114 + return err
115 + }
116 + // the server assumes if we ack the broadcast, we've updated
117 + // our levels. Hence the order.

the server will wait ExchangeTimeout and then kill the session, quicker to send a nack? it seems a serious problem on the client, not sure how you expect to recover without user intervention?

118 + err = sess.proto.WriteMessage(protocol.AckMsg{"ack"})

[Samuele Pedroni (pedronis)]

        state := uint32(Disconnected)
        levels, err := levelmap.NewLevelMap()
        if err != nil {
                return nil, err
        }

the error path there is not covered fwiw

[Samuele Pedroni (pedronis)]

71 + c.Check(err, IsNil)

should this one and similar be Assert?

[John Lenton (chipaca)]

Yes! yes, it should send a nack. Fixing.

The error path gets covered further down the pipeline, when there is a levelmap that can actually return non-nil error.

I use c.Assert on the error when I need to dereference the result (and in this case I don't). Otherwise I'd rather let it go ahead.

[lp:~chipaca/ubuntu-push/prepersistance]

62. By John Lenton on 2014-02-12

send nack on levelmap set error

63. By John Lenton on 2014-02-12

nak, not nack

[Samuele Pedroni (pedronis)]

looks good

review: Approve

=== modified file 'client/session/levelmap/levelmap.go'
--- client/session/levelmap/levelmap.go	2014-01-30 10:46:21 +0000
+++ client/session/levelmap/levelmap.go	2014-02-12 13:24:35 +0000
@@ -24,23 +24,24 @@
 
 type LevelMap interface {
 	// Set() (re)sets the given level to the given value.
-	Set(level string, top int64)
+	Set(level string, top int64) error
 	// GetAll() returns a "simple" map of the current levels.
-	GetAll() map[string]int64
+	GetAll() (map[string]int64, error)
 }
 
 type mapLevelMap map[string]int64
 
-func (m *mapLevelMap) Set(level string, top int64) {
+func (m *mapLevelMap) Set(level string, top int64) error {
 	(*m)[level] = top
+	return nil
 }
-func (m *mapLevelMap) GetAll() map[string]int64 {
-	return map[string]int64(*m)
+func (m *mapLevelMap) GetAll() (map[string]int64, error) {
+	return map[string]int64(*m), nil
 }
 
 var _ LevelMap = &mapLevelMap{}
 
 // default constructor
-func NewLevelMap() LevelMap {
-	return &mapLevelMap{}
+func NewLevelMap() (LevelMap, error) {
+	return &mapLevelMap{}, nil
 }
 
=== modified file 'client/session/levelmap/levelmap_test.go'
--- client/session/levelmap/levelmap_test.go	2014-01-30 10:46:21 +0000
+++ client/session/levelmap/levelmap_test.go	2014-02-12 13:24:35 +0000
@@ -23,19 +23,33 @@
 
 func TestLevelMap(t *testing.T) { TestingT(t) }
 
-type lmSuite struct{}
+type lmSuite struct {
+	constructor func() (LevelMap, error)
+}
 
 var _ = Suite(&lmSuite{})
 
-func (cs *lmSuite) TestAllTheThings(c *C) {
+func (s *lmSuite) SetUpSuite(c *C) {
+	s.constructor = NewLevelMap
+}
+
+func (s *lmSuite) TestAllTheThings(c *C) {
+	var err error
+	var lm LevelMap
 	// checks NewLevelMap returns a LevelMap
-	var lm LevelMap = NewLevelMap()
+	lm, err = s.constructor()
+	// and that it works
+	c.Assert(err, IsNil)
 	// setting a couple of things, sets them
-	lm.Set("this", 12)
-	lm.Set("that", 42)
-	c.Check(lm.GetAll(), DeepEquals, map[string]int64{"this": 12, "that": 42})
+	c.Check(lm.Set("this", 12), IsNil)
+	c.Check(lm.Set("that", 42), IsNil)
+	all, err := lm.GetAll()
+	c.Check(err, IsNil)
+	c.Check(all, DeepEquals, map[string]int64{"this": 12, "that": 42})
 	// re-setting one of them, resets it
-	lm.Set("this", 999)
-	c.Check(lm.GetAll(), DeepEquals, map[string]int64{"this": 999, "that": 42})
+	c.Check(lm.Set("this", 999), IsNil)
+	all, err = lm.GetAll()
+	c.Check(err, IsNil)
+	c.Check(all, DeepEquals, map[string]int64{"this": 999, "that": 42})
 	// huzzah
 }
 
=== modified file 'client/session/session.go'
--- client/session/session.go	2014-02-05 13:23:53 +0000
+++ client/session/session.go	2014-02-12 13:24:35 +0000
@@ -80,13 +80,17 @@
 func NewSession(serverAddr string, pem []byte, exchangeTimeout time.Duration,
 	deviceId string, log logger.Logger) (*ClientSession, error) {
 	state := uint32(Disconnected)
+	levels, err := levelmap.NewLevelMap()
+	if err != nil {
+		return nil, err
+	}
 	sess := &ClientSession{
 		ExchangeTimeout: exchangeTimeout,
 		ServerAddr:      serverAddr,
 		DeviceId:        deviceId,
 		Log:             log,
 		Protocolator:    protocol.NewProtocol0,
-		Levels:          levelmap.NewLevelMap(),
+		Levels:          levels,
 		TLS:             &tls.Config{InsecureSkipVerify: true}, // XXX
 		stateP:          &state,
 	}
@@ -164,7 +168,16 @@
 
 // handle "broadcast" messages
 func (sess *ClientSession) handleBroadcast(bcast *serverMsg) error {
-	err := sess.proto.WriteMessage(protocol.AckMsg{"ack"})
+	err := sess.Levels.Set(bcast.ChanId, bcast.TopLevel)
+	if err != nil {
+		sess.setState(Error)
+		sess.Log.Errorf("unable to set level: %v", err)
+		sess.proto.WriteMessage(protocol.AckMsg{"nak"})
+		return err
+	}
+	// the server assumes if we ack the broadcast, we've updated
+	// our levels. Hence the order.
+	err = sess.proto.WriteMessage(protocol.AckMsg{"ack"})
 	if err != nil {
 		sess.setState(Error)
 		sess.Log.Errorf("unable to ack broadcast: %s", err)
@@ -175,7 +188,6 @@
 	if bcast.ChanId == protocol.SystemChannelId {
 		// the system channel id, the only one we care about for now
 		sess.Log.Debugf("sending it over")
-		sess.Levels.Set(bcast.ChanId, bcast.TopLevel)
 		sess.MsgCh <- &Notification{}
 		sess.Log.Debugf("sent it over")
 	} else {
@@ -228,10 +240,16 @@
 	}
 	proto := sess.Protocolator(conn)
 	proto.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+	levels, err := sess.Levels.GetAll()
+	if err != nil {
+		sess.setState(Error)
+		sess.Log.Errorf("unable to start: get levels: %v", err)
+		return err
+	}
 	err = proto.WriteMessage(protocol.ConnectMsg{
 		Type:     "connect",
 		DeviceId: sess.DeviceId,
-		Levels:   sess.Levels.GetAll(),
+		Levels:   levels,
 	})
 	if err != nil {
 		sess.setState(Error)
 
=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-02-05 18:23:47 +0000
+++ client/session/session_test.go	2014-02-12 13:24:35 +0000
@@ -156,6 +156,14 @@
 	return nil
 }
 
+// brokenLevelMap is a LevelMap that always breaks
+type brokenLevelMap struct{}
+
+func (*brokenLevelMap) Set(string, int64) error           { return errors.New("broken.") }
+func (*brokenLevelMap) GetAll() (map[string]int64, error) { return nil, errors.New("broken.") }
+
+/////
+
 func (cs *clientSessionSuite) SetUpTest(c *C) {
 	cs.log = helpers.NewTestLogger(c, "debug")
 }
@@ -365,7 +373,9 @@
 	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})
+	levels, err := s.sess.Levels.GetAll()
+	c.Check(err, IsNil)
+	c.Check(levels, DeepEquals, map[string]int64{"0": 2})
 }
 
 func (s *msgSuite) TestHandleBroadcastBadAckWrite(c *C) {
@@ -401,6 +411,27 @@
 	c.Check(len(s.sess.MsgCh), Equals, 0)
 }
 
+func (s *msgSuite) TestHandleBroadcastWrongBrokenLevelmap(c *C) {
+	s.sess.Levels = &brokenLevelMap{}
+	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) }()
+	s.upCh <- nil // ack ok
+	// start returns with error
+	c.Check(<-s.errCh, Not(Equals), nil)
+	// no message sent out
+	c.Check(len(s.sess.MsgCh), Equals, 0)
+	// and nak'ed it
+	c.Check(len(s.downCh), Equals, 1)
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"nak"})
+}
+
 /****************************************************************
   loop() tests
 ****************************************************************/
@@ -488,6 +519,26 @@
 	c.Check(sess.State(), Equals, Error)
 }
 
+func (cs *clientSessionSuite) TestStartFailsIfGetLevelsFails(c *C) {
+	sess, err := NewSession("", nil, 0, "wah", cs.log)
+	c.Assert(err, IsNil)
+	sess.Levels = &brokenLevelMap{}
+	sess.Connection = &testConn{Name: "TestStartConnectMessageFails"}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.start()
+	}()
+
+	c.Check(takeNext(downCh), Equals, "deadline 0")
+	err = <-errCh
+	c.Check(err, ErrorMatches, "broken.")
+}
+
 func (cs *clientSessionSuite) TestStartConnectMessageFails(c *C) {
 	sess, err := NewSession("", nil, 0, "wah", cs.log)
 	c.Assert(err, IsNil)
@@ -788,7 +839,9 @@
 	// ...get bubbled up,
 	c.Check(<-sess.MsgCh, NotNil)
 	// and their TopLevel remembered
-	c.Check(sess.Levels.GetAll(), DeepEquals, map[string]int64{"0": 2})
+	levels, err := sess.Levels.GetAll()
+	c.Check(err, IsNil)
+	c.Check(levels, DeepEquals, map[string]int64{"0": 2})
 
 	// and ping still work even after that.
 	c.Check(takeNext(downCh), Equals, "deadline 110ms")