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

[Samuele Pedroni]

add support for unicast channels to the pending store

add support for unicast channels to the pending store

add the concept/support for unicast chan identifiers, plus some settling/clarifications

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

143. By Samuele Pedroni on 2014-05-01

Merged unicast-preps-2 into unicast-store.

[John Lenton (chipaca)]

Good.

review: Approve

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

144. By Samuele Pedroni on 2014-05-02

rename the function DropByMsgId -> FilterOutByMsgId

=== modified file 'server/store/inmemory.go'
--- server/store/inmemory.go	2014-05-02 15:13:24 +0000
+++ server/store/inmemory.go	2014-05-02 15:13:24 +0000
@@ -24,16 +24,11 @@
 	"launchpad.net/ubuntu-push/protocol"
 )
 
-// one stored notification
-type notification struct {
-	protocol.Notification
-	expiration time.Time
-}
-
 // one stored channel
 type channel struct {
 	topLevel      int64
-	notifications []notification
+	notifications []protocol.Notification
+	expirations   []time.Time
 }
 
 // InMemoryPendingStore is a basic in-memory pending notification store.
@@ -56,22 +51,34 @@
 	return InternalChannelId(""), ErrUnknownChannel
 }
 
-func (sto *InMemoryPendingStore) AppendToChannel(chanId InternalChannelId, notificationPayload json.RawMessage, expiration time.Time) error {
+func (sto *InMemoryPendingStore) appendToChannel(chanId InternalChannelId, newNotification protocol.Notification, inc int64, expiration time.Time) error {
 	sto.lock.Lock()
 	defer sto.lock.Unlock()
 	prev := sto.store[chanId]
 	if prev == nil {
 		prev = &channel{}
 	}
-	prev.topLevel++
-	prev.notifications = append(prev.notifications, notification{
-		Notification: protocol.Notification{Payload: notificationPayload},
-		expiration:   expiration,
-	})
+	prev.topLevel += inc
+	prev.notifications = append(prev.notifications, newNotification)
+	prev.expirations = append(prev.expirations, expiration)
 	sto.store[chanId] = prev
 	return nil
 }
 
+func (sto *InMemoryPendingStore) AppendToChannel(chanId InternalChannelId, notificationPayload json.RawMessage, expiration time.Time) error {
+	newNotification := protocol.Notification{Payload: notificationPayload}
+	return sto.appendToChannel(chanId, newNotification, 1, expiration)
+}
+
+func (sto *InMemoryPendingStore) AppendToUnicastChannel(chanId InternalChannelId, appId string, notificationPayload json.RawMessage, msgId string, expiration time.Time) error {
+	newNotification := protocol.Notification{
+		Payload: notificationPayload,
+		AppId:   appId,
+		MsgId:   msgId,
+	}
+	return sto.appendToChannel(chanId, newNotification, 0, expiration)
+}
+
 func (sto *InMemoryPendingStore) GetChannelSnapshot(chanId InternalChannelId) (int64, []protocol.Notification, error) {
 	sto.lock.Lock()
 	defer sto.lock.Unlock()
@@ -82,13 +89,18 @@
 	topLevel := channel.topLevel
 	n := len(channel.notifications)
 	res := make([]protocol.Notification, 0, n)
+	exps := make([]time.Time, 0, n)
 	now := time.Now()
-	for _, notification := range channel.notifications {
-		if notification.expiration.Before(now) {
+	for i, expiration := range channel.expirations {
+		if expiration.Before(now) {
 			continue
 		}
-		res = append(res, notification.Notification)
+		res = append(res, channel.notifications[i])
+		exps = append(exps, expiration)
 	}
+	// store as well
+	channel.notifications = res
+	channel.expirations = exps
 	return topLevel, res, nil
 }
 
@@ -96,5 +108,25 @@
 	// ignored
 }
 
+func (sto *InMemoryPendingStore) DropByMsgId(chanId InternalChannelId, targets []protocol.Notification) error {
+	sto.lock.Lock()
+	defer sto.lock.Unlock()
+	channel, ok := sto.store[chanId]
+	if !ok {
+		return nil
+	}
+	expById := make(map[string]time.Time, len(channel.notifications))
+	for i, notif := range channel.notifications {
+		expById[notif.MsgId] = channel.expirations[i]
+	}
+	channel.notifications = FilterOutByMsgId(channel.notifications, targets)
+	exps := make([]time.Time, len(channel.notifications))
+	for i, notif := range channel.notifications {
+		exps[i] = expById[notif.MsgId]
+	}
+	channel.expirations = exps
+	return nil
+}
+
 // sanity check we implement the interface
 var _ PendingStore = (*InMemoryPendingStore)(nil)

=== modified file 'server/store/inmemory_test.go'
--- server/store/inmemory_test.go	2014-05-02 15:13:24 +0000
+++ server/store/inmemory_test.go	2014-05-02 15:13:24 +0000
@@ -67,6 +67,32 @@
 	c.Check(res, DeepEquals, help.Ns(notification1, notification2))
 }
 
+func (s *inMemorySuite) TestAppendToUnicastChannelAndGetChannelSnapshot(c *C) {
+	sto := NewInMemoryPendingStore()
+
+	chanId := UnicastInternalChannelId("user", "dev1")
+	notification1 := json.RawMessage(`{"a":1}`)
+	notification2 := json.RawMessage(`{"b":2}`)
+	app1 := "app1"
+	app2 := "app2"
+	msg1 := "msg1"
+	msg2 := "msg2"
+
+	muchLater := time.Now().Add(time.Minute)
+
+	err := sto.AppendToUnicastChannel(chanId, app1, notification1, msg1, muchLater)
+	c.Assert(err, IsNil)
+	err = sto.AppendToUnicastChannel(chanId, app2, notification2, msg2, muchLater)
+	c.Assert(err, IsNil)
+	top, res, err := sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+	c.Check(res, DeepEquals, []protocol.Notification{
+		protocol.Notification{Payload: notification1, AppId: app1, MsgId: msg1},
+		protocol.Notification{Payload: notification2, AppId: app2, MsgId: msg2},
+	})
+	c.Check(top, Equals, int64(0))
+}
+
 func (s *inMemorySuite) TestAppendToChannelAndGetChannelSnapshotWithExpiration(c *C) {
 	sto := NewInMemoryPendingStore()
 
@@ -86,3 +112,43 @@
 	c.Check(top, Equals, int64(2))
 	c.Check(res, DeepEquals, help.Ns(notification1))
 }
+
+func (s *inMemorySuite) TestDropByMsgId(c *C) {
+	sto := NewInMemoryPendingStore()
+
+	chanId := UnicastInternalChannelId("user", "dev2")
+
+	// nothing to do is fine
+	err := sto.DropByMsgId(chanId, nil)
+	c.Assert(err, IsNil)
+
+	notification1 := json.RawMessage(`{"a":1}`)
+	notification2 := json.RawMessage(`{"b":2}`)
+	notification3 := json.RawMessage(`{"a":2}`)
+	app1 := "app1"
+	app2 := "app2"
+	msg1 := "msg1"
+	msg2 := "msg2"
+	msg3 := "msg3"
+
+	muchLater := time.Now().Add(time.Minute)
+
+	err = sto.AppendToUnicastChannel(chanId, app1, notification1, msg1, muchLater)
+	c.Assert(err, IsNil)
+	err = sto.AppendToUnicastChannel(chanId, app2, notification2, msg2, muchLater)
+	c.Assert(err, IsNil)
+	err = sto.AppendToUnicastChannel(chanId, app1, notification3, msg3, muchLater)
+	c.Assert(err, IsNil)
+	_, res, err := sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+
+	err = sto.DropByMsgId(chanId, res[:2])
+	c.Assert(err, IsNil)
+
+	_, res, err = sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+	c.Check(res, HasLen, 1)
+	c.Check(res, DeepEquals, []protocol.Notification{
+		protocol.Notification{Payload: notification3, AppId: app1, MsgId: msg3},
+	})
+}

=== modified file 'server/store/store.go'
--- server/store/store.go	2014-05-02 15:13:24 +0000
+++ server/store/store.go	2014-05-02 15:13:24 +0000
@@ -21,6 +21,8 @@
 	"encoding/hex"
 	"encoding/json"
 	"errors"
+	"fmt"
+	"strings"
 	"time"
 
 	"launchpad.net/ubuntu-push/protocol"
@@ -28,6 +30,27 @@
 
 type InternalChannelId string
 
+// BroadcastChannel returns whether the id represents a broadcast channel.
+func (icid InternalChannelId) BroadcastChannel() bool {
+	marker := icid[0]
+	return marker == 'B' || marker == '0'
+}
+
+// UnicastChannel returns whether the id represents a unicast channel.
+func (icid InternalChannelId) UnicastChannel() bool {
+	marker := icid[0]
+	return marker == 'U'
+}
+
+// UnicastUserAndDevice returns the user and device ids of a unicast channel.
+func (icid InternalChannelId) UnicastUserAndDevice() (userId, deviceId string) {
+	if !icid.UnicastChannel() {
+		panic("UnicastUserAndDevice is for unicast channels")
+	}
+	parts := strings.SplitN(string(icid)[1:], ":", 2)
+	return parts[0], parts[1]
+}
+
 var ErrUnknownChannel = errors.New("unknown channel name")
 var ErrFull = errors.New("channel is full")
 var ErrExpected128BitsHexRepr = errors.New("expected 128 bits hex repr")
@@ -38,7 +61,10 @@
 	if chanId == SystemInternalChannelId {
 		return "0"
 	}
-	panic("general InternalChannelIdToHex not implemeted yet")
+	if !chanId.BroadcastChannel() {
+		panic("InternalChannelIdToHex is for broadcast channels")
+	}
+	return string(chanId)[1:]
 }
 
 var zero128 [16]byte
@@ -60,7 +86,14 @@
 	if idbytes == zero128 {
 		return SystemInternalChannelId, nil
 	}
-	return InternalChannelId(idbytes[:]), nil
+	// mark with B(roadcast) prefix
+	s := "B" + hexRepr
+	return InternalChannelId(s), nil
+}
+
+// UnicastInternalChannelId builds a channel id for the userId, deviceId pair.
+func UnicastInternalChannelId(userId, deviceId string) InternalChannelId {
+	return InternalChannelId(fmt.Sprintf("U%s:%s", userId, deviceId))
 }
 
 // PendingStore let store notifications into channels.
@@ -70,9 +103,45 @@
 	GetInternalChannelId(name string) (InternalChannelId, error)
 	// AppendToChannel appends a notification to the channel.
 	AppendToChannel(chanId InternalChannelId, notification json.RawMessage, expiration time.Time) error
+	// AppendToUnicastChannel appends a notification to the unicast channel.
 	// GetChannelSnapshot gets all the current notifications and
+	AppendToUnicastChannel(chanId InternalChannelId, appId string, notification json.RawMessage, msgId string, expiration time.Time) error
 	// current top level in the channel.
 	GetChannelSnapshot(chanId InternalChannelId) (topLevel int64, notifications []protocol.Notification, err error)
+	// DropByMsgId drops notifications from a unicast channel based on message ids.
+	DropByMsgId(chanId InternalChannelId, targets []protocol.Notification) error
 	// Close is to be called when done with the store.
 	Close()
 }
+
+// FilterOutByMsgId returns the notifications from orig whose msg id is not
+// mentioned in targets.
+func FilterOutByMsgId(orig, targets []protocol.Notification) []protocol.Notification {
+	n := len(orig)
+	t := len(targets)
+	// common case, removing the continuous head
+	if t > 0 && n >= t {
+		if targets[0].MsgId == orig[0].MsgId {
+			for i := t - 1; i >= 0; i-- {
+				if i == 0 {
+					return orig[t:]
+				}
+				if targets[i].MsgId != orig[i].MsgId {
+					break
+				}
+			}
+		}
+	}
+	// slow way
+	ids := make(map[string]bool, t)
+	for _, target := range targets {
+		ids[target.MsgId] = true
+	}
+	acc := make([]protocol.Notification, 0, n)
+	for _, notif := range orig {
+		if !ids[notif.MsgId] {
+			acc = append(acc, notif)
+		}
+	}
+	return acc
+}

=== modified file 'server/store/store_test.go'
--- server/store/store_test.go	2014-02-10 23:19:08 +0000
+++ server/store/store_test.go	2014-05-02 15:13:24 +0000
@@ -33,6 +33,8 @@
 
 func (s *storeSuite) TestInternalChannelIdToHex(c *C) {
 	c.Check(InternalChannelIdToHex(SystemInternalChannelId), Equals, protocol.SystemChannelId)
+	c.Check(InternalChannelIdToHex(InternalChannelId("Bf1c9bf7096084cb2a154979ce00c7f50")), Equals, "f1c9bf7096084cb2a154979ce00c7f50")
+	c.Check(func() { InternalChannelIdToHex(InternalChannelId("U")) }, PanicMatches, "InternalChannelIdToHex is for broadcast channels")
 }
 
 func (s *storeSuite) TestHexToInternalChannelId(c *C) {
@@ -42,9 +44,11 @@
 	i1, err := HexToInternalChannelId("00000000000000000000000000000000")
 	c.Check(err, IsNil)
 	c.Check(i1, Equals, SystemInternalChannelId)
+	c.Check(i1.BroadcastChannel(), Equals, true)
 	i2, err := HexToInternalChannelId("f1c9bf7096084cb2a154979ce00c7f50")
 	c.Check(err, IsNil)
-	c.Check(i2, Equals, InternalChannelId("\xf1\xc9\xbf\x70\x96\x08\x4c\xb2\xa1\x54\x97\x9c\xe0\x0c\x7f\x50"))
+	c.Check(i2.BroadcastChannel(), Equals, true)
+	c.Check(i2, Equals, InternalChannelId("Bf1c9bf7096084cb2a154979ce00c7f50"))
 	_, err = HexToInternalChannelId("01")
 	c.Check(err, Equals, ErrExpected128BitsHexRepr)
 	_, err = HexToInternalChannelId("abceddddddddddddddddzeeeeeeeeeee")
@@ -52,3 +56,44 @@
 	_, err = HexToInternalChannelId("f1c9bf7096084cb2a154979ce00c7f50ff")
 	c.Check(err, Equals, ErrExpected128BitsHexRepr)
 }
+
+func (s *storeSuite) TestUnicastInternalChannelId(c *C) {
+	chanId := UnicastInternalChannelId("user1", "dev2")
+	c.Check(chanId.BroadcastChannel(), Equals, false)
+	c.Check(chanId.UnicastChannel(), Equals, true)
+	u, d := chanId.UnicastUserAndDevice()
+	c.Check(u, Equals, "user1")
+	c.Check(d, Equals, "dev2")
+	c.Check(func() { SystemInternalChannelId.UnicastUserAndDevice() }, PanicMatches, "UnicastUserAndDevice is for unicast channels")
+}
+
+func (s *storeSuite) TestFilterOutByMsgId(c *C) {
+	orig := []protocol.Notification{
+		protocol.Notification{MsgId: "a"},
+		protocol.Notification{MsgId: "b"},
+		protocol.Notification{MsgId: "c"},
+		protocol.Notification{MsgId: "d"},
+	}
+	// removing the continuous head
+	res := FilterOutByMsgId(orig, orig[:3])
+	c.Check(res, DeepEquals, orig[3:])
+
+	// random removal
+	res = FilterOutByMsgId(orig, orig[1:2])
+	c.Check(res, DeepEquals, []protocol.Notification{
+		protocol.Notification{MsgId: "a"},
+		protocol.Notification{MsgId: "c"},
+		protocol.Notification{MsgId: "d"},
+	})
+
+	// looks like removing the continuous head, but it isn't
+	res = FilterOutByMsgId(orig, []protocol.Notification{
+		protocol.Notification{MsgId: "a"},
+		protocol.Notification{MsgId: "c"},
+		protocol.Notification{MsgId: "d"},
+	})
+	c.Check(res, DeepEquals, []protocol.Notification{
+		protocol.Notification{MsgId: "b"},
+	})
+
+}