Merge lp:~pedronis/ubuntu-push/freelance-exchanges into lp:ubuntu-push

[Samuele Pedroni]

exchanges don't require to be so tied to a particular broker

with a bit of reorg exchanges don't require to be so tied to a particular broker

[John Lenton (chipaca)]

Only issues I'd have with this one are fixed in the next one. Good job.

review: Approve

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-01-14 15:35:20 +0000
+++ server/broker/broker.go	2014-01-24 07:56:28 +0000
@@ -44,6 +44,9 @@
 	Acked(BrokerSession) error
 }
 
+// LevelsMap is the type for holding levels for session.
+type LevelsMap map[store.InternalChannelId]int64
+
 // BrokerSession holds broker session state.
 type BrokerSession interface {
 	// SessionChannel returns the session control channel
@@ -51,6 +54,10 @@
 	SessionChannel() <-chan Exchange
 	// DeviceId returns the device id string.
 	DeviceId() string
+	// Levels returns the current channel levels for the session
+	Levels() LevelsMap
+	// ExchangeScratchArea returns the scratch area for exchanges.
+	ExchangeScratchArea() *ExchangesScratchArea
 }
 
 // Session aborted error.

=== added file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	1970-01-01 00:00:00 +0000
+++ server/broker/exchanges.go	2014-01-24 07:56:28 +0000
@@ -0,0 +1,74 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package broker
+
+import (
+	"encoding/json"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/store"
+	// "log"
+)
+
+// Exchanges
+
+// Scratch area for exchanges, sessions should one of these.
+type ExchangesScratchArea struct {
+	broadcastMsg protocol.BroadcastMsg
+	ackMsg       protocol.AckMsg
+}
+
+// BroadcastExchange leads a session through delivering a BROADCAST.
+// For simplicity its fully public.
+type BroadcastExchange struct {
+	ChanId               store.InternalChannelId
+	TopLevel             int64
+	NotificationPayloads []json.RawMessage
+}
+
+func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
+	c := int64(len(payloads))
+	delta := topLevel - clientLevel
+	if delta < c {
+		return payloads[c-delta:]
+	} else {
+		return payloads
+	}
+}
+
+// Prepare session for a BROADCAST.
+func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
+	scratchArea := sess.ExchangeScratchArea()
+	scratchArea.broadcastMsg.Type = "broadcast"
+	clientLevel := sess.Levels()[sbe.ChanId]
+	payloads := filterByLevel(clientLevel, sbe.TopLevel, sbe.NotificationPayloads)
+	// xxx need an AppId as well, later
+	scratchArea.broadcastMsg.ChanId = store.InternalChannelIdToHex(sbe.ChanId)
+	scratchArea.broadcastMsg.TopLevel = sbe.TopLevel
+	scratchArea.broadcastMsg.Payloads = payloads
+	return &scratchArea.broadcastMsg, &scratchArea.ackMsg, nil
+}
+
+// Acked deals with an ACK for a BROADCAST.
+func (sbe *BroadcastExchange) Acked(sess BrokerSession) error {
+	scratchArea := sess.ExchangeScratchArea()
+	if scratchArea.ackMsg.Type != "ack" {
+		return &ErrAbort{"expected ACK message"}
+	}
+	// update levels
+	sess.Levels()[sbe.ChanId] = sbe.TopLevel
+	return nil
+}

=== added file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	1970-01-01 00:00:00 +0000
+++ server/broker/exchanges_test.go	2014-01-24 07:56:28 +0000
@@ -0,0 +1,123 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package broker
+
+import (
+	"encoding/json"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/server/store"
+	// "log"
+)
+
+type exchangesSuite struct{}
+
+var _ = Suite(&exchangesSuite{})
+
+func (s *exchangesSuite) TestBroadcastExchange(c *C) {
+	sess := &simpleBrokerSession{
+		levels: map[store.InternalChannelId]int64{},
+	}
+	exchg := &BroadcastExchange{
+		ChanId:   store.SystemInternalChannelId,
+		TopLevel: 3,
+		NotificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"x"}`),
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	inMsg, outMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(inMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"x"},{"a":"y"}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess)
+	c.Assert(err, IsNil)
+	c.Check(sess.levels[store.SystemInternalChannelId], Equals, int64(3))
+}
+
+func (s *exchangesSuite) TestBroadcastExchangeAckMismatch(c *C) {
+	sess := &simpleBrokerSession{
+		levels: map[store.InternalChannelId]int64{},
+	}
+	exchg := &BroadcastExchange{
+		ChanId:   store.SystemInternalChannelId,
+		TopLevel: 3,
+		NotificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	inMsg, outMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(inMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
+	err = json.Unmarshal([]byte(`{}`), outMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess)
+	c.Assert(err, Not(IsNil))
+	c.Check(sess.levels[store.SystemInternalChannelId], Equals, int64(0))
+}
+
+func (s *exchangesSuite) TestFilterByLevel(c *C) {
+	payloads := []json.RawMessage{
+		json.RawMessage(`{"a": 3}`),
+		json.RawMessage(`{"a": 4}`),
+		json.RawMessage(`{"a": 5}`),
+	}
+	res := filterByLevel(5, 5, payloads)
+	c.Check(len(res), Equals, 0)
+	res = filterByLevel(4, 5, payloads)
+	c.Check(len(res), Equals, 1)
+	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 5}`))
+	res = filterByLevel(3, 5, payloads)
+	c.Check(len(res), Equals, 2)
+	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 4}`))
+	res = filterByLevel(2, 5, payloads)
+	c.Check(len(res), Equals, 3)
+	res = filterByLevel(1, 5, payloads)
+	c.Check(len(res), Equals, 3)
+}
+
+func (s *exchangesSuite) TestBroadcastExchangeFilterByLevel(c *C) {
+	sess := &simpleBrokerSession{
+		levels: map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 2,
+		},
+	}
+	exchg := &BroadcastExchange{
+		ChanId:   store.SystemInternalChannelId,
+		TopLevel: 3,
+		NotificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"x"}`),
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	inMsg, outMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(inMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess)
+	c.Assert(err, IsNil)
+}

=== modified file 'server/broker/simple.go'
--- server/broker/simple.go	2014-01-14 15:35:20 +0000
+++ server/broker/simple.go	2014-01-24 07:56:28 +0000
@@ -17,7 +17,6 @@
 package broker
 
 import (
-	"encoding/json"
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/store"
@@ -48,10 +47,9 @@
 	deviceId   string
 	done       chan bool
 	exchanges  chan Exchange
-	levels     map[store.InternalChannelId]int64
+	levels     LevelsMap
 	// for exchanges
-	broadcastMsg protocol.BroadcastMsg
-	ackMsg       protocol.AckMsg
+	exchgScratch ExchangesScratchArea
 }
 
 type deliveryKind int
@@ -74,6 +72,14 @@
 	return sess.deviceId
 }
 
+func (sess *simpleBrokerSession) Levels() LevelsMap {
+	return sess.levels
+}
+
+func (sess *simpleBrokerSession) ExchangeScratchArea() *ExchangesScratchArea {
+	return &sess.exchgScratch
+}
+
 // NewSimpleBroker makes a new SimpleBroker.
 func NewSimpleBroker(sto store.PendingStore, cfg BrokerConfig, logger logger.Logger) *SimpleBroker {
 	sessionCh := make(chan *simpleBrokerSession, cfg.BrokerQueueSize())
@@ -126,10 +132,10 @@
 		}
 		clientLevel := sess.levels[chanId]
 		if clientLevel != topLevel {
-			broadcastExchg := &simpleBroadcastExchange{
-				chanId:               chanId,
-				topLevel:             topLevel,
-				notificationPayloads: payloads,
+			broadcastExchg := &BroadcastExchange{
+				ChanId:               chanId,
+				TopLevel:             topLevel,
+				NotificationPayloads: payloads,
 			}
 			sess.exchanges <- broadcastExchg
 		}
@@ -198,10 +204,10 @@
 					b.logger.Errorf("unsuccessful broadcast, get channel snapshot for %v: %v", delivery.chanId, err)
 					continue Loop
 				}
-				broadcastExchg := &simpleBroadcastExchange{
-					chanId:               delivery.chanId,
-					topLevel:             topLevel,
-					notificationPayloads: payloads,
+				broadcastExchg := &BroadcastExchange{
+					ChanId:               delivery.chanId,
+					TopLevel:             topLevel,
+					NotificationPayloads: payloads,
 				}
 				for _, sess := range b.registry {
 					sess.exchanges <- broadcastExchg
@@ -218,43 +224,3 @@
 		chanId: chanId,
 	}
 }
-
-// Exchanges
-
-type simpleBroadcastExchange struct {
-	chanId               store.InternalChannelId
-	topLevel             int64
-	notificationPayloads []json.RawMessage
-}
-
-func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
-	c := int64(len(payloads))
-	delta := topLevel - clientLevel
-	if delta < c {
-		return payloads[c-delta:]
-	} else {
-		return payloads
-	}
-}
-
-func (sbe *simpleBroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
-	simpleSess := sess.(*simpleBrokerSession)
-	simpleSess.broadcastMsg.Type = "broadcast"
-	clientLevel := simpleSess.levels[sbe.chanId]
-	payloads := filterByLevel(clientLevel, sbe.topLevel, sbe.notificationPayloads)
-	// xxx need an AppId as well, later
-	simpleSess.broadcastMsg.ChanId = store.InternalChannelIdToHex(sbe.chanId)
-	simpleSess.broadcastMsg.TopLevel = sbe.topLevel
-	simpleSess.broadcastMsg.Payloads = payloads
-	return &simpleSess.broadcastMsg, &simpleSess.ackMsg, nil
-}
-
-func (sbe *simpleBroadcastExchange) Acked(sess BrokerSession) error {
-	simpleSess := sess.(*simpleBrokerSession)
-	if simpleSess.ackMsg.Type != "ack" {
-		return &ErrAbort{"expected ACK message"}
-	}
-	// update levels
-	simpleSess.levels[sbe.chanId] = sbe.topLevel
-	return nil
-}

=== modified file 'server/broker/simple_test.go'
--- server/broker/simple_test.go	2014-01-14 15:35:20 +0000
+++ server/broker/simple_test.go	2014-01-24 07:56:28 +0000
@@ -69,9 +69,9 @@
 	c.Assert(err, IsNil)
 	c.Assert(b.registry["dev-1"], Equals, sess)
 	c.Assert(sess.DeviceId(), Equals, "dev-1")
-	c.Check(sess.(*simpleBrokerSession).levels, DeepEquals, map[store.InternalChannelId]int64{
+	c.Check(sess.Levels(), DeepEquals, LevelsMap(map[store.InternalChannelId]int64{
 		store.SystemInternalChannelId: 5,
-	})
+	}))
 	b.Unregister(sess)
 	// just to make sure the unregister was processed
 	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""})
@@ -99,10 +99,10 @@
 	b.feedPending(sess)
 	c.Assert(len(sess.exchanges), Equals, 1)
 	exchg1 := <-sess.exchanges
-	c.Check(exchg1, DeepEquals, &simpleBroadcastExchange{
-		chanId:               store.SystemInternalChannelId,
-		topLevel:             1,
-		notificationPayloads: []json.RawMessage{notification1},
+	c.Check(exchg1, DeepEquals, &BroadcastExchange{
+		ChanId:               store.SystemInternalChannelId,
+		TopLevel:             1,
+		NotificationPayloads: []json.RawMessage{notification1},
 	})
 }
 
@@ -163,101 +163,6 @@
 	c.Check(b.registry["dev-1"], Equals, sess2)
 }
 
-func (s *simpleSuite) TestBroadcastExchange(c *C) {
-	sess := &simpleBrokerSession{
-		levels: map[store.InternalChannelId]int64{},
-	}
-	exchg := &simpleBroadcastExchange{
-		chanId:   store.SystemInternalChannelId,
-		topLevel: 3,
-		notificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"x"}`),
-			json.RawMessage(`{"a":"y"}`),
-		},
-	}
-	inMsg, outMsg, err := exchg.Prepare(sess)
-	c.Assert(err, IsNil)
-	// check
-	marshalled, err := json.Marshal(inMsg)
-	c.Assert(err, IsNil)
-	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"x"},{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
-	c.Assert(err, IsNil)
-	err = exchg.Acked(sess)
-	c.Assert(err, IsNil)
-	c.Check(sess.levels[store.SystemInternalChannelId], Equals, int64(3))
-}
-
-func (s *simpleSuite) TestBroadcastExchangeAckMismatch(c *C) {
-	sess := &simpleBrokerSession{
-		levels: map[store.InternalChannelId]int64{},
-	}
-	exchg := &simpleBroadcastExchange{
-		chanId:   store.SystemInternalChannelId,
-		topLevel: 3,
-		notificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"y"}`),
-		},
-	}
-	inMsg, outMsg, err := exchg.Prepare(sess)
-	c.Assert(err, IsNil)
-	// check
-	marshalled, err := json.Marshal(inMsg)
-	c.Assert(err, IsNil)
-	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{}`), outMsg)
-	c.Assert(err, IsNil)
-	err = exchg.Acked(sess)
-	c.Assert(err, Not(IsNil))
-	c.Check(sess.levels[store.SystemInternalChannelId], Equals, int64(0))
-}
-
-func (s *simpleSuite) TestFilterByLevel(c *C) {
-	payloads := []json.RawMessage{
-		json.RawMessage(`{"a": 3}`),
-		json.RawMessage(`{"a": 4}`),
-		json.RawMessage(`{"a": 5}`),
-	}
-	res := filterByLevel(5, 5, payloads)
-	c.Check(len(res), Equals, 0)
-	res = filterByLevel(4, 5, payloads)
-	c.Check(len(res), Equals, 1)
-	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 5}`))
-	res = filterByLevel(3, 5, payloads)
-	c.Check(len(res), Equals, 2)
-	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 4}`))
-	res = filterByLevel(2, 5, payloads)
-	c.Check(len(res), Equals, 3)
-	res = filterByLevel(1, 5, payloads)
-	c.Check(len(res), Equals, 3)
-}
-
-func (s *simpleSuite) TestBroadcastExchangeFilterByLevel(c *C) {
-	sess := &simpleBrokerSession{
-		levels: map[store.InternalChannelId]int64{
-			store.SystemInternalChannelId: 2,
-		},
-	}
-	exchg := &simpleBroadcastExchange{
-		chanId:   store.SystemInternalChannelId,
-		topLevel: 3,
-		notificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"x"}`),
-			json.RawMessage(`{"a":"y"}`),
-		},
-	}
-	inMsg, outMsg, err := exchg.Prepare(sess)
-	c.Assert(err, IsNil)
-	// check
-	marshalled, err := json.Marshal(inMsg)
-	c.Assert(err, IsNil)
-	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
-	c.Assert(err, IsNil)
-	err = exchg.Acked(sess)
-	c.Assert(err, IsNil)
-}
-
 func (s *simpleSuite) TestBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	notification1 := json.RawMessage(`{"m": "M"}`)
@@ -274,20 +179,20 @@
 	case <-time.After(5 * time.Second):
 		c.Fatal("taking too long to get broadcast exchange")
 	case exchg1 := <-sess1.SessionChannel():
-		c.Check(exchg1, DeepEquals, &simpleBroadcastExchange{
-			chanId:               store.SystemInternalChannelId,
-			topLevel:             1,
-			notificationPayloads: []json.RawMessage{notification1},
+		c.Check(exchg1, DeepEquals, &BroadcastExchange{
+			ChanId:               store.SystemInternalChannelId,
+			TopLevel:             1,
+			NotificationPayloads: []json.RawMessage{notification1},
 		})
 	}
 	select {
 	case <-time.After(5 * time.Second):
 		c.Fatal("taking too long to get broadcast exchange")
 	case exchg2 := <-sess2.SessionChannel():
-		c.Check(exchg2, DeepEquals, &simpleBroadcastExchange{
-			chanId:               store.SystemInternalChannelId,
-			topLevel:             1,
-			notificationPayloads: []json.RawMessage{notification1},
+		c.Check(exchg2, DeepEquals, &BroadcastExchange{
+			ChanId:               store.SystemInternalChannelId,
+			TopLevel:             1,
+			NotificationPayloads: []json.RawMessage{notification1},
 		})
 	}
 }

=== modified file 'server/session/session_test.go'
--- server/session/session_test.go	2014-01-15 15:54:20 +0000
+++ server/session/session_test.go	2014-01-24 07:56:28 +0000
@@ -136,6 +136,14 @@
 	return tbs.deviceId
 }
 
+func (tbs *testBrokerSession) Levels() broker.LevelsMap {
+	return nil
+}
+
+func (tbs *testBrokerSession) ExchangeScratchArea() *broker.ExchangesScratchArea {
+	return nil
+}
+
 func (tb *testBroker) Register(connect *protocol.ConnectMsg) (broker.BrokerSession, error) {
 	tb.registration <- "register " + connect.DeviceId
 	return &testBrokerSession{connect.DeviceId, nil}, tb.err