Merge lp:~pedronis/ubuntu-push/unicast-feed-pending-and-reorg into lp:ubuntu-push/automatic

[Samuele Pedroni]

support feeding pending unicasts at session start, together with/through some related reorg

support feeding pending unicasts at session start, together with/through some related reorg

[John Lenton (chipaca)]

review: Approve

=== modified file 'server/acceptance/suites/unicast.go'
--- server/acceptance/suites/unicast.go	2014-05-02 11:20:55 +0000
+++ server/acceptance/suites/unicast.go	2014-05-15 15:14:22 +0000
@@ -94,3 +94,24 @@
 	c.Check(len(errCh1), Equals, 0)
 	c.Check(len(errCh2), Equals, 0)
 }
+
+func (s *UnicastAcceptanceSuite) TestUnicastPending(c *C) {
+	// send unicast that will be pending
+	userId, auth := s.associatedAuth("DEV1")
+	got, err := s.PostRequest("/notify", &api.Unicast{
+		UserId:   userId,
+		DeviceId: "DEV1",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 42}`),
+	})
+	c.Assert(err, IsNil)
+	c.Assert(got, Matches, ".*ok.*")
+
+	// get pending on connect
+	events, errCh, stop := s.StartClientAuth(c, "DEV1", nil, auth)
+	c.Check(NextEvent(events, errCh), Equals, `unicast app:app1 payload:{"a":42};`)
+	stop()
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
 
=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-05-01 18:56:07 +0000
+++ server/broker/broker.go	2014-05-15 15:14:22 +0000
@@ -87,6 +87,10 @@
 	Get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error)
 	// DropByMsgId drops notifications from the channel chanId by message id.
 	DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error
+	// Feed feeds exchange into the session.
+	Feed(Exchange)
+	// InternalChannelId() returns the channel id corresponding to the session.
+	InternalChannelId() store.InternalChannelId
 }
 
 // Session aborted error.
 
=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-05-01 18:56:07 +0000
+++ server/broker/exchanges.go	2014-05-15 15:14:22 +0000
@@ -19,6 +19,7 @@
 import (
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/store"
@@ -33,6 +34,10 @@
 	ackMsg           protocol.AckMsg
 }
 
+type BaseExchange struct {
+	Timestamp time.Time
+}
+
 // BroadcastExchange leads a session through delivering a BROADCAST.
 // For simplicity it is fully public.
 type BroadcastExchange struct {
@@ -40,6 +45,7 @@
 	TopLevel      int64
 	Notifications []protocol.Notification
 	Decoded       []map[string]interface{}
+	BaseExchange
 }
 
 // check interface already here
@@ -140,7 +146,9 @@
 // UnicastExchange leads a session through delivering a NOTIFICATIONS message.
 // For simplicity it is fully public.
 type UnicastExchange struct {
-	ChanId store.InternalChannelId
+	ChanId   store.InternalChannelId
+	CachedOk bool
+	BaseExchange
 }
 
 // check interface already here
@@ -148,10 +156,13 @@
 
 // Prepare session for a NOTIFICATIONS.
 func (sue *UnicastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
-	_, notifs, err := sess.Get(sue.ChanId, false)
+	_, notifs, err := sess.Get(sue.ChanId, sue.CachedOk)
 	if err != nil {
 		return nil, nil, err
 	}
+	if len(notifs) == 0 {
+		return nil, nil, ErrNop
+	}
 	scratchArea := sess.ExchangeScratchArea()
 	scratchArea.notificationsMsg.Reset()
 	scratchArea.notificationsMsg.Type = "notifications"
@@ -171,3 +182,28 @@
 	}
 	return nil
 }
+
+// FeedPending feeds exchanges covering pending notifications into the session.
+func FeedPending(sess BrokerSession) error {
+	// find relevant channels, for now only system
+	channels := []store.InternalChannelId{store.SystemInternalChannelId}
+	for _, chanId := range channels {
+		topLevel, notifications, err := sess.Get(chanId, true)
+		if err != nil {
+			// next broadcast will try again
+			continue
+		}
+		clientLevel := sess.Levels()[chanId]
+		if clientLevel != topLevel {
+			broadcastExchg := &BroadcastExchange{
+				ChanId:        chanId,
+				TopLevel:      topLevel,
+				Notifications: notifications,
+			}
+			broadcastExchg.Init()
+			sess.Feed(broadcastExchg)
+		}
+	}
+	sess.Feed(&UnicastExchange{ChanId: sess.InternalChannelId(), CachedOk: true})
+	return nil
+}
 
=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-05-01 18:56:07 +0000
+++ server/broker/exchanges_test.go	2014-05-15 15:14:22 +0000
@@ -295,7 +295,7 @@
 			return nil
 		},
 	}
-	exchg := &broker.UnicastExchange{chanId1}
+	exchg := &broker.UnicastExchange{ChanId: chanId1, CachedOk: false}
 	outMsg, inMsg, err := exchg.Prepare(sess)
 	c.Assert(err, IsNil)
 	// check
@@ -311,7 +311,7 @@
 }
 
 func (s *exchangesSuite) TestUnicastExchangeAckMismatch(c *C) {
-	notifs := []protocol.Notification{}
+	notifs := []protocol.Notification{protocol.Notification{}}
 	dropped := make(chan []protocol.Notification, 2)
 	sess := &testing.TestBrokerSession{
 		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
@@ -344,8 +344,22 @@
 	c.Assert(err, Equals, fail)
 }
 
+func (s *exchangesSuite) TestUnicastExchangeCachedOkNop(c *C) {
+	chanId1 := store.UnicastInternalChannelId("u1", "d1")
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			c.Check(chanId, Equals, chanId1)
+			c.Check(cachedOk, Equals, true)
+			return 0, nil, nil
+		},
+	}
+	exchg := &broker.UnicastExchange{ChanId: chanId1, CachedOk: true}
+	_, _, err := exchg.Prepare(sess)
+	c.Assert(err, Equals, broker.ErrNop)
+}
+
 func (s *exchangesSuite) TestUnicastExchangeErrorOnAcked(c *C) {
-	notifs := []protocol.Notification{}
+	notifs := []protocol.Notification{protocol.Notification{}}
 	fail := errors.New("fail")
 	sess := &testing.TestBrokerSession{
 		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
@@ -363,3 +377,79 @@
 	err = exchg.Acked(sess, true)
 	c.Assert(err, Equals, fail)
 }
+
+func (s *exchangesSuite) TestFeedPending(c *C) {
+	bcast1 := json.RawMessage(`{"m": "M"}`)
+	decoded1 := map[string]interface{}{"m": "M"}
+	sess := &testing.TestBrokerSession{
+		Exchanges: make(chan broker.Exchange, 5),
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			switch chanId {
+			case store.SystemInternalChannelId:
+				return 1, help.Ns(bcast1), nil
+			default:
+				return 0, nil, nil
+			}
+		},
+	}
+	err := broker.FeedPending(sess)
+	c.Assert(err, IsNil)
+	c.Assert(len(sess.Exchanges), Equals, 2)
+	exchg1 := <-sess.Exchanges
+	c.Check(exchg1, DeepEquals, &broker.BroadcastExchange{
+		ChanId:        store.SystemInternalChannelId,
+		TopLevel:      1,
+		Notifications: help.Ns(bcast1),
+		Decoded:       []map[string]interface{}{decoded1},
+	})
+	exchg2 := <-sess.Exchanges
+	c.Check(exchg2, DeepEquals, &broker.UnicastExchange{
+		ChanId:   sess.InternalChannelId(),
+		CachedOk: true,
+	})
+}
+
+func (s *exchangesSuite) TestFeedPendingSystemChanNop(c *C) {
+	bcast1 := json.RawMessage(`{"m": "M"}`)
+	sess := &testing.TestBrokerSession{
+		LevelsMap: map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 1,
+		},
+		Exchanges: make(chan broker.Exchange, 5),
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			switch chanId {
+			case store.SystemInternalChannelId:
+				return 1, help.Ns(bcast1), nil
+			default:
+				return 0, nil, nil
+			}
+		},
+	}
+	err := broker.FeedPending(sess)
+	c.Assert(err, IsNil)
+	c.Check(len(sess.Exchanges), Equals, 1)
+	exchg1 := <-sess.Exchanges
+	c.Check(exchg1, FitsTypeOf, &broker.UnicastExchange{})
+}
+
+func (s *exchangesSuite) TestFeedPendingSystemChanFail(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap: map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 1,
+		},
+		Exchanges: make(chan broker.Exchange, 5),
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			switch chanId {
+			case store.SystemInternalChannelId:
+				return 0, nil, errors.New("fail")
+			default:
+				return 0, nil, nil
+			}
+		},
+	}
+	err := broker.FeedPending(sess)
+	c.Assert(err, IsNil)
+	c.Check(len(sess.Exchanges), Equals, 1)
+	exchg1 := <-sess.Exchanges
+	c.Check(exchg1, FitsTypeOf, &broker.UnicastExchange{})
+}
 
=== modified file 'server/broker/simple/simple.go'
--- server/broker/simple/simple.go	2014-05-01 18:56:07 +0000
+++ server/broker/simple/simple.go	2014-05-15 15:14:22 +0000
@@ -103,6 +103,14 @@
 	return sess.broker.drop(chanId, targets)
 }
 
+func (sess *simpleBrokerSession) Feed(exchg broker.Exchange) {
+	sess.exchanges <- exchg
+}
+
+func (sess *simpleBrokerSession) InternalChannelId() store.InternalChannelId {
+	return store.UnicastInternalChannelId(sess.deviceId, sess.deviceId)
+}
+
 // NewSimpleBroker makes a new SimpleBroker.
 func NewSimpleBroker(sto store.PendingStore, cfg broker.BrokerConfig, logger logger.Logger) *SimpleBroker {
 	sessionCh := make(chan *simpleBrokerSession, cfg.BrokerQueueSize())
@@ -150,30 +158,6 @@
 	return b.running
 }
 
-func (b *SimpleBroker) feedPending(sess *simpleBrokerSession) error {
-	// find relevant channels, for now only system
-	channels := []store.InternalChannelId{store.SystemInternalChannelId}
-	for _, chanId := range channels {
-		topLevel, notifications, err := b.sto.GetChannelSnapshot(chanId)
-		if err != nil {
-			// next broadcast will try again
-			b.logger.Errorf("unsuccessful feed pending, get channel snapshot for %v: %v", chanId, err)
-			continue
-		}
-		clientLevel := sess.levels[chanId]
-		if clientLevel != topLevel {
-			broadcastExchg := &broker.BroadcastExchange{
-				ChanId:        chanId,
-				TopLevel:      topLevel,
-				Notifications: notifications,
-			}
-			broadcastExchg.Init()
-			sess.exchanges <- broadcastExchg
-		}
-	}
-	return nil
-}
-
 // Register registers a session with the broker. It feeds the session
 // pending notifications as well.
 func (b *SimpleBroker) Register(connect *protocol.ConnectMsg, sessionId string) (broker.BrokerSession, error) {
@@ -205,7 +189,7 @@
 	}
 	b.sessionCh <- sess
 	<-sess.done
-	err = b.feedPending(sess)
+	err = broker.FeedPending(sess)
 	if err != nil {
 		return nil, err
 	}
@@ -281,7 +265,7 @@
 				_, devId := chanId.UnicastUserAndDevice()
 				sess := b.registry[devId]
 				if sess != nil {
-					sess.exchanges <- &broker.UnicastExchange{chanId}
+					sess.exchanges <- &broker.UnicastExchange{ChanId: chanId, CachedOk: false}
 				}
 			}
 		}
 
=== modified file 'server/broker/simple/simple_test.go'
--- server/broker/simple/simple_test.go	2014-04-29 15:29:04 +0000
+++ server/broker/simple/simple_test.go	2014-05-15 15:14:22 +0000
@@ -17,16 +17,12 @@
 package simple
 
 import (
-	"encoding/json"
 	stdtesting "testing"
-	"time"
 
 	. "launchpad.net/gocheck"
 
-	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/broker/testing"
 	"launchpad.net/ubuntu-push/server/store"
-	help "launchpad.net/ubuntu-push/testing"
 )
 
 func TestSimple(t *stdtesting.T) { TestingT(t) }
@@ -45,39 +41,7 @@
 	c.Check(b.sto, Equals, sto)
 }
 
-func (s *simpleSuite) TestFeedPending(c *C) {
-	sto := store.NewInMemoryPendingStore()
-	muchLater := time.Now().Add(10 * time.Minute)
-	notification1 := json.RawMessage(`{"m": "M"}`)
-	decoded1 := map[string]interface{}{"m": "M"}
-	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
-	b := NewSimpleBroker(sto, testBrokerConfig, nil)
-	sess := &simpleBrokerSession{
-		exchanges: make(chan broker.Exchange, 1),
-	}
-	b.feedPending(sess)
-	c.Assert(len(sess.exchanges), Equals, 1)
-	exchg1 := <-sess.exchanges
-	c.Check(exchg1, DeepEquals, &broker.BroadcastExchange{
-		ChanId:        store.SystemInternalChannelId,
-		TopLevel:      1,
-		Notifications: help.Ns(notification1),
-		Decoded:       []map[string]interface{}{decoded1},
-	})
-}
-
-func (s *simpleSuite) TestFeedPendingNop(c *C) {
-	sto := store.NewInMemoryPendingStore()
-	muchLater := time.Now().Add(10 * time.Minute)
-	notification1 := json.RawMessage(`{"m": "M"}`)
-	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
-	b := NewSimpleBroker(sto, testBrokerConfig, nil)
-	sess := &simpleBrokerSession{
-		exchanges: make(chan broker.Exchange, 1),
-		levels: map[store.InternalChannelId]int64{
-			store.SystemInternalChannelId: 1,
-		},
-	}
-	b.feedPending(sess)
-	c.Assert(len(sess.exchanges), Equals, 0)
+func (s *simpleSuite) TestSessionInternalChannelId(c *C) {
+	sess := &simpleBrokerSession{deviceId: "dev21"}
+	c.Check(sess.InternalChannelId(), Equals, store.UnicastInternalChannelId("dev21", "dev21"))
 }
 
=== modified file 'server/broker/testing/impls.go'
--- server/broker/testing/impls.go	2014-05-01 18:56:07 +0000
+++ server/broker/testing/impls.go	2014-05-15 15:14:22 +0000
@@ -68,6 +68,14 @@
 	return tbs.DoDropByMsgId(chanId, targets)
 }
 
+func (tbs *TestBrokerSession) Feed(exchg broker.Exchange) {
+	tbs.Exchanges <- exchg
+}
+
+func (tbs *TestBrokerSession) InternalChannelId() store.InternalChannelId {
+	return store.UnicastInternalChannelId(tbs.DeviceId, tbs.DeviceId)
+}
+
 // Test implementation of BrokerConfig.
 type TestBrokerConfig struct {
 	ConfigSessionQueueSize uint
 
=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-05-01 18:56:07 +0000
+++ server/broker/testsuite/suite.go	2014-05-15 15:14:22 +0000
@@ -143,7 +143,7 @@
 	defer b.Stop()
 	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
-	c.Check(len(sess.SessionChannel()), Equals, 1)
+	c.Check(len(sess.SessionChannel()), Equals, 2)
 }
 
 func (s *CommonBrokerSuite) TestRegistrationFeedPendingError(c *C) {
@@ -154,7 +154,12 @@
 	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
 	// but
-	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful feed pending, get channel snapshot for 0: get channel snapshot fail\n")
+	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful, get channel snapshot for 0 \\(cachedOk=true\\): get channel snapshot fail\n")
+}
+
+func clearOfPending(c *C, sess broker.BrokerSession) {
+	c.Assert(len(sess.SessionChannel()) >= 1, Equals, true)
+	<-sess.SessionChannel()
 }
 
 func (s *CommonBrokerSuite) TestRegistrationLastWins(c *C) {
@@ -164,6 +169,7 @@
 	defer b.Stop()
 	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
+	clearOfPending(c, sess1)
 	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s2")
 	c.Assert(err, IsNil)
 	// previous session got signaled by sending nil on its channel
@@ -194,8 +200,10 @@
 	defer b.Stop()
 	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
+	clearOfPending(c, sess1)
 	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-2"}, "s2")
 	c.Assert(err, IsNil)
+	clearOfPending(c, sess2)
 	// add notification to channel *after* the registrations
 	muchLater := time.Now().Add(10 * time.Minute)
 	sto.AppendToChannel(store.SystemInternalChannelId, notification1, muchLater)
@@ -250,8 +258,9 @@
 	b := s.MakeBroker(sto, testBrokerConfig, s.testlog)
 	b.Start()
 	defer b.Stop()
-	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
+	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
+	clearOfPending(c, sess)
 	b.Broadcast(store.SystemInternalChannelId)
 	select {
 	case <-time.After(5 * time.Second):
@@ -272,8 +281,10 @@
 	defer b.Stop()
 	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev1"}, "s1")
 	c.Assert(err, IsNil)
+	clearOfPending(c, sess1)
 	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev2"}, "s2")
 	c.Assert(err, IsNil)
+	clearOfPending(c, sess2)
 	// add notification to channel *after* the registrations
 	muchLater := time.Now().Add(10 * time.Minute)
 	sto.AppendToUnicastChannel(chanId1, "app1", notification1, "msg1", muchLater)
@@ -338,3 +349,20 @@
 	c.Assert(err, ErrorMatches, "drop fail")
 	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful, drop from channel Udev3:dev3: drop fail\n")
 }
+
+func (s *CommonBrokerSuite) TestSessionFeed(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	b := s.MakeBroker(sto, testBrokerConfig, nil)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev3"}, "s1")
+	c.Assert(err, IsNil)
+	clearOfPending(c, sess1)
+	bcast := &broker.BroadcastExchange{ChanId: store.SystemInternalChannelId, TopLevel: 99}
+	sess1.Feed(bcast)
+	c.Check(s.RevealBroadcastExchange(<-sess1.SessionChannel()), DeepEquals, bcast)
+
+	ucast := &broker.UnicastExchange{ChanId: store.UnicastInternalChannelId("dev21", "dev21"), CachedOk: true}
+	sess1.Feed(ucast)
+	c.Check(s.RevealUnicastExchange(<-sess1.SessionChannel()), DeepEquals, ucast)
+}