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

[Samuele Pedroni]

unicast in broker and introduce UnicastExchange, some reorg

unicast in broker and introduce UnicastExchange, some reorg

for later branches:

- deal with pending unicast messages
- support splitting large NOTIFICATIONS messages

[John Lenton (chipaca)]

review: Approve

[lp:~pedronis/ubuntu-push/unicast-broker]

146. By Samuele Pedroni on 2014-05-02

Merged unicast-store into unicast-broker.

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-05-02 15:13:38 +0000
+++ protocol/messages.go	2014-05-02 15:13:38 +0000
@@ -154,6 +154,17 @@
 	Notifications []Notification
 }
 
+// Reset resets the splitting state if the message storage is to be
+// reused.
+func (m *NotificationsMsg) Reset() {
+	// xxx
+}
+
+func (m *NotificationsMsg) Split() bool {
+	// xxx
+	return true
+}
+
 // A single unicast notification
 type Notification struct {
 	AppId string `json:"A"`

=== modified file 'server/api/handlers_test.go'
--- server/api/handlers_test.go	2014-05-02 15:13:38 +0000
+++ server/api/handlers_test.go	2014-05-02 15:13:38 +0000
@@ -150,6 +150,10 @@
 	cbsend.payloads = protocol.ExtractPayloads(notifications)
 }
 
+func (cbsend *checkBrokerSending) Unicast(chanIds ...store.InternalChannelId) {
+	// xxx later
+}
+
 func (s *handlersSuite) TestDoBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	bsend := &checkBrokerSending{store: sto}
@@ -269,6 +273,10 @@
 	bsend.chanId <- chanId
 }
 
+func (bsend testBrokerSending) Unicast(chanIds ...store.InternalChannelId) {
+	// xxx later
+}
+
 func (s *handlersSuite) TestRespondsToBasicSystemBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-04-17 15:42:27 +0000
+++ server/broker/broker.go	2014-05-02 15:13:38 +0000
@@ -39,6 +39,8 @@
 type BrokerSending interface {
 	// Broadcast channel.
 	Broadcast(chanId store.InternalChannelId)
+	// Unicast over channels.
+	Unicast(chanIds ...store.InternalChannelId)
 }
 
 // Exchange leads the session through performing an exchange, typically delivery.
@@ -81,6 +83,10 @@
 	Levels() LevelsMap
 	// ExchangeScratchArea returns the scratch area for exchanges.
 	ExchangeScratchArea() *ExchangesScratchArea
+	// Get gets the content of the channel with chanId.
+	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
 }
 
 // Session aborted error.

=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-05-02 15:13:38 +0000
+++ server/broker/exchanges.go	2014-05-02 15:13:38 +0000
@@ -28,8 +28,9 @@
 
 // Scratch area for exchanges, sessions should hold one of these.
 type ExchangesScratchArea struct {
-	broadcastMsg protocol.BroadcastMsg
-	ackMsg       protocol.AckMsg
+	broadcastMsg     protocol.BroadcastMsg
+	notificationsMsg protocol.NotificationsMsg
+	ackMsg           protocol.AckMsg
 }
 
 // BroadcastExchange leads a session through delivering a BROADCAST.
@@ -135,3 +136,38 @@
 func (cbe *ConnMetaExchange) Acked(sess BrokerSession, done bool) error {
 	panic("Acked should not get invoked on ConnMetaExchange")
 }
+
+// UnicastExchange leads a session through delivering a NOTIFICATIONS message.
+// For simplicity it is fully public.
+type UnicastExchange struct {
+	ChanId store.InternalChannelId
+}
+
+// check interface already here
+var _ Exchange = (*UnicastExchange)(nil)
+
+// 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)
+	if err != nil {
+		return nil, nil, err
+	}
+	scratchArea := sess.ExchangeScratchArea()
+	scratchArea.notificationsMsg.Reset()
+	scratchArea.notificationsMsg.Type = "notifications"
+	scratchArea.notificationsMsg.Notifications = notifs
+	return &scratchArea.notificationsMsg, &scratchArea.ackMsg, nil
+}
+
+// Acked deals with an ACK for a NOTIFICATIONS.
+func (sue *UnicastExchange) Acked(sess BrokerSession, done bool) error {
+	scratchArea := sess.ExchangeScratchArea()
+	if scratchArea.ackMsg.Type != "ack" {
+		return &ErrAbort{"expected ACK message"}
+	}
+	err := sess.DropByMsgId(sue.ChanId, scratchArea.notificationsMsg.Notifications)
+	if err != nil {
+		return err
+	}
+	return nil
+}

=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-05-02 15:13:38 +0000
+++ server/broker/exchanges_test.go	2014-05-02 15:13:38 +0000
@@ -18,6 +18,7 @@
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	stdtesting "testing"
@@ -266,3 +267,99 @@
 
 	c.Check(func() { cbe.Acked(nil, true) }, PanicMatches, "Acked should not get invoked on ConnMetaExchange")
 }
+
+func (s *exchangesSuite) TestUnicastExchange(c *C) {
+	chanId1 := store.UnicastInternalChannelId("u1", "d1")
+	notifs := []protocol.Notification{
+		protocol.Notification{
+			MsgId:   "msg1",
+			AppId:   "app1",
+			Payload: json.RawMessage(`{"m": 1}`),
+		},
+		protocol.Notification{
+			MsgId:   "msg2",
+			AppId:   "app2",
+			Payload: json.RawMessage(`{"m": 2}`),
+		},
+	}
+	dropped := make(chan []protocol.Notification, 2)
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			c.Check(chanId, Equals, chanId1)
+			c.Check(cachedOk, Equals, false)
+			return 0, notifs, nil
+		},
+		DoDropByMsgId: func(chanId store.InternalChannelId, targets []protocol.Notification) error {
+			c.Check(chanId, Equals, chanId1)
+			dropped <- targets
+			return nil
+		},
+	}
+	exchg := &broker.UnicastExchange{chanId1}
+	outMsg, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(outMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"notifications","Notifications":[{"A":"app1","M":"msg1","P":{"m":1}},{"A":"app2","M":"msg2","P":{"m":2}}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, IsNil)
+	c.Check(dropped, HasLen, 1)
+	c.Check(<-dropped, DeepEquals, notifs)
+}
+
+func (s *exchangesSuite) TestUnicastExchangeAckMismatch(c *C) {
+	notifs := []protocol.Notification{}
+	dropped := make(chan []protocol.Notification, 2)
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			return 0, notifs, nil
+		},
+		DoDropByMsgId: func(chanId store.InternalChannelId, targets []protocol.Notification) error {
+			dropped <- targets
+			return nil
+		},
+	}
+	exchg := &broker.UnicastExchange{}
+	_, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	err = json.Unmarshal([]byte(`{}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, Not(IsNil))
+	c.Check(dropped, HasLen, 0)
+}
+
+func (s *exchangesSuite) TestUnicastExchangeErrorOnPrepare(c *C) {
+	fail := errors.New("fail")
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			return 0, nil, fail
+		},
+	}
+	exchg := &broker.UnicastExchange{}
+	_, _, err := exchg.Prepare(sess)
+	c.Assert(err, Equals, fail)
+}
+
+func (s *exchangesSuite) TestUnicastExchangeErrorOnAcked(c *C) {
+	notifs := []protocol.Notification{}
+	fail := errors.New("fail")
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			return 0, notifs, nil
+		},
+		DoDropByMsgId: func(chanId store.InternalChannelId, targets []protocol.Notification) error {
+			return fail
+		},
+	}
+	exchg := &broker.UnicastExchange{}
+	_, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, Equals, fail)
+}

=== modified file 'server/broker/simple/simple.go'
--- server/broker/simple/simple.go	2014-05-02 15:13:38 +0000
+++ server/broker/simple/simple.go	2014-05-02 15:13:38 +0000
@@ -46,6 +46,7 @@
 
 // simpleBrokerSession represents a session in the broker.
 type simpleBrokerSession struct {
+	broker       *SimpleBroker
 	registered   bool
 	deviceId     string
 	model        string
@@ -61,6 +62,7 @@
 
 const (
 	broadcastDelivery deliveryKind = iota
+	unicastDelivery
 )
 
 // delivery holds all the information to request a delivery
@@ -93,6 +95,14 @@
 	return &sess.exchgScratch
 }
 
+func (sess *simpleBrokerSession) Get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+	return sess.broker.get(chanId, cachedOk)
+}
+
+func (sess *simpleBrokerSession) DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	return sess.broker.drop(chanId, targets)
+}
+
 // NewSimpleBroker makes a new SimpleBroker.
 func NewSimpleBroker(sto store.PendingStore, cfg broker.BrokerConfig, logger logger.Logger) *SimpleBroker {
 	sessionCh := make(chan *simpleBrokerSession, cfg.BrokerQueueSize())
@@ -185,6 +195,7 @@
 		levels[id] = v
 	}
 	sess := &simpleBrokerSession{
+		broker:       b,
 		deviceId:     connect.DeviceId,
 		model:        model,
 		imageChannel: imageChannel,
@@ -207,6 +218,24 @@
 	b.sessionCh <- sess
 }
 
+func (b *SimpleBroker) get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+	topLevel, notifications, err := b.sto.GetChannelSnapshot(chanId)
+	if err != nil {
+		b.logger.Errorf("unsuccessful, get channel snapshot for %v (cachedOk=%v): %v", chanId, cachedOk, err)
+	}
+	return topLevel, notifications, err
+
+}
+
+func (b *SimpleBroker) drop(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	err := b.sto.DropByMsgId(chanId, targets)
+	if err != nil {
+		b.logger.Errorf("unsuccessful, drop from channel %v: %v", chanId, err)
+	}
+	return err
+
+}
+
 // run runs the agent logic of the broker.
 func (b *SimpleBroker) run() {
 Loop:
@@ -233,10 +262,9 @@
 		case delivery := <-b.deliveryCh:
 			switch delivery.kind {
 			case broadcastDelivery:
-				topLevel, notifications, err := b.sto.GetChannelSnapshot(delivery.chanId)
+				topLevel, notifications, err := b.get(delivery.chanId, false)
 				if err != nil {
 					// next broadcast will try again
-					b.logger.Errorf("unsuccessful broadcast, get channel snapshot for %v: %v", delivery.chanId, err)
 					continue Loop
 				}
 				broadcastExchg := &broker.BroadcastExchange{
@@ -248,6 +276,13 @@
 				for _, sess := range b.registry {
 					sess.exchanges <- broadcastExchg
 				}
+			case unicastDelivery:
+				chanId := delivery.chanId
+				_, devId := chanId.UnicastUserAndDevice()
+				sess := b.registry[devId]
+				if sess != nil {
+					sess.exchanges <- &broker.UnicastExchange{chanId}
+				}
 			}
 		}
 	}
@@ -260,3 +295,13 @@
 		chanId: chanId,
 	}
 }
+
+// Unicast requests unicast for the channels.
+func (b *SimpleBroker) Unicast(chanIds ...store.InternalChannelId) {
+	for _, chanId := range chanIds {
+		b.deliveryCh <- &delivery{
+			kind:   unicastDelivery,
+			chanId: chanId,
+		}
+	}
+}

=== modified file 'server/broker/simple/suite_test.go'
--- server/broker/simple/suite_test.go	2014-02-10 23:19:08 +0000
+++ server/broker/simple/suite_test.go	2014-05-02 15:13:38 +0000
@@ -42,4 +42,7 @@
 	RevealBroadcastExchange: func(exchg broker.Exchange) *broker.BroadcastExchange {
 		return exchg.(*broker.BroadcastExchange)
 	},
+	RevealUnicastExchange: func(exchg broker.Exchange) *broker.UnicastExchange {
+		return exchg.(*broker.UnicastExchange)
+	},
 }})

=== modified file 'server/broker/testing/impls.go'
--- server/broker/testing/impls.go	2014-04-03 14:31:10 +0000
+++ server/broker/testing/impls.go	2014-05-02 15:13:38 +0000
@@ -18,7 +18,9 @@
 package testing
 
 import (
+	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/broker"
+	"launchpad.net/ubuntu-push/server/store"
 )
 
 // Test implementation of BrokerSession.
@@ -29,6 +31,9 @@
 	Exchanges    chan broker.Exchange
 	LevelsMap    broker.LevelsMap
 	exchgScratch broker.ExchangesScratchArea
+	// hooks
+	DoGet         func(store.InternalChannelId, bool) (int64, []protocol.Notification, error)
+	DoDropByMsgId func(store.InternalChannelId, []protocol.Notification) error
 }
 
 func (tbs *TestBrokerSession) DeviceIdentifier() string {
@@ -55,6 +60,14 @@
 	return &tbs.exchgScratch
 }
 
+func (tbs *TestBrokerSession) Get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+	return tbs.DoGet(chanId, cachedOk)
+}
+
+func (tbs *TestBrokerSession) DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	return tbs.DoDropByMsgId(chanId, targets)
+}
+
 // Test implementation of BrokerConfig.
 type TestBrokerConfig struct {
 	ConfigSessionQueueSize uint

=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-05-02 15:13:38 +0000
+++ server/broker/testsuite/suite.go	2014-05-02 15:13:38 +0000
@@ -50,6 +50,8 @@
 	RevealSession func(broker.Broker, string) broker.BrokerSession
 	// Let us get to a broker.BroadcastExchange from an Exchange.
 	RevealBroadcastExchange func(broker.Exchange) *broker.BroadcastExchange
+	// Let us get to a broker.UnicastExchange from an Exchange.
+	RevealUnicastExchange func(broker.Exchange) *broker.UnicastExchange
 	// private
 	testlog *help.TestLogger
 }
@@ -235,6 +237,10 @@
 	return 0, nil, nil
 }
 
+func (sto *testFailingStore) DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	return errors.New("drop fail")
+}
+
 func (s *CommonBrokerSuite) TestBroadcastFail(c *C) {
 	logged := make(chan bool, 1)
 	s.testlog.SetLogEventCb(func(string) {
@@ -252,5 +258,83 @@
 		c.Fatal("taking too long to log error")
 	case <-logged:
 	}
-	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful broadcast, get channel snapshot for 0: get channel snapshot fail\n")
+	c.Check(s.testlog.Captured(), Matches, "ERROR.*: get channel snapshot fail\n")
+}
+
+func (s *CommonBrokerSuite) TestUnicast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M1"}`)
+	notification2 := json.RawMessage(`{"m": "M2"}`)
+	chanId1 := store.UnicastInternalChannelId("dev1", "dev1")
+	chanId2 := store.UnicastInternalChannelId("dev2", "dev2")
+	b := s.MakeBroker(sto, testBrokerConfig, nil)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev1"}, "s1")
+	c.Assert(err, IsNil)
+	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev2"}, "s2")
+	c.Assert(err, IsNil)
+	// add notification to channel *after* the registrations
+	muchLater := time.Now().Add(10 * time.Minute)
+	sto.AppendToUnicastChannel(chanId1, "app1", notification1, "msg1", muchLater)
+	sto.AppendToUnicastChannel(chanId2, "app2", notification2, "msg2", muchLater)
+	b.Unicast(chanId2, chanId1)
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get unicast exchange")
+	case exchg1 := <-sess1.SessionChannel():
+		u1 := s.RevealUnicastExchange(exchg1)
+		c.Check(u1.ChanId, Equals, chanId1)
+	}
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get unicast exchange")
+	case exchg2 := <-sess2.SessionChannel():
+		u2 := s.RevealUnicastExchange(exchg2)
+		c.Check(u2.ChanId, Equals, chanId2)
+	}
+}
+
+func (s *CommonBrokerSuite) TestGetAndDrop(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M1"}`)
+	chanId1 := store.UnicastInternalChannelId("dev3", "dev3")
+	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)
+	muchLater := time.Now().Add(10 * time.Minute)
+	sto.AppendToUnicastChannel(chanId1, "app1", notification1, "msg1", muchLater)
+	_, expected, err := sto.GetChannelSnapshot(chanId1)
+	c.Assert(err, IsNil)
+	_, notifs, err := sess1.Get(chanId1, false)
+	c.Check(notifs, HasLen, 1)
+	c.Check(notifs, DeepEquals, expected)
+	err = sess1.DropByMsgId(chanId1, notifs)
+	c.Assert(err, IsNil)
+	_, notifs, err = sess1.Get(chanId1, true)
+	c.Check(notifs, HasLen, 0)
+	_, expected, err = sto.GetChannelSnapshot(chanId1)
+	c.Assert(err, IsNil)
+	c.Check(expected, HasLen, 0)
+
+}
+
+func (s *CommonBrokerSuite) TestGetAndDropErrors(c *C) {
+	chanId1 := store.UnicastInternalChannelId("dev3", "dev3")
+	sto := &testFailingStore{countdownToFail: 1}
+	b := s.MakeBroker(sto, testBrokerConfig, s.testlog)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev3"}, "s1")
+	c.Assert(err, IsNil)
+	_, _, err = sess1.Get(chanId1, false)
+	c.Assert(err, ErrorMatches, "get channel snapshot fail")
+	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful, get channel snapshot for Udev3:dev3 \\(cachedOk=false\\): get channel snapshot fail\n")
+	s.testlog.ResetCapture()
+
+	err = sess1.DropByMsgId(chanId1, nil)
+	c.Assert(err, ErrorMatches, "drop fail")
+	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful, drop from channel Udev3:dev3: drop fail\n")
 }