Merge lp:~pedronis/ubuntu-push/use-info into lp:ubuntu-push/automatic

[Samuele Pedroni]

use image informations sent by the client to filter the system channel broadcasts

use image informations sent by the client to filter the system channel broadcasts,

assume there will be an entry in them with key "IMAGE-CHANNEL/DEVICE"

[John Lenton (chipaca)]

WOo! :)

review: Approve

[Ubuntu One Auto Pilot (otto-pilot)]

The attempt to merge lp:~pedronis/ubuntu-push/use-info into lp:ubuntu-push/automatic failed. Below is the output from the failed tests.

mkdir -p /mnt/tarmac/cache/ubuntu-push-automatic/go-ws/bin
mkdir -p /mnt/tarmac/cache/ubuntu-push-automatic/go-ws/pkg
go get -u launchpad.net/godeps
go get -d -u launchpad.net/gocheck launchpad.net/go-dbus/v1 launchpad.net/go-xdg/v0 code.google.com/p/gosqlite/sqlite3
/mnt/tarmac/cache/ubuntu-push-automatic/go-ws/bin/godeps -u dependencies.tsv
"/mnt/tarmac/cache/ubuntu-push-automatic/go-ws/src/launchpad.net/gocheck" now at <email address hidden>
go install launchpad.net/gocheck launchpad.net/go-dbus/v1 launchpad.net/go-xdg/v0 code.google.com/p/gosqlite/sqlite3
go test launchpad.net/ubuntu-push launchpad.net/ubuntu-push/bus launchpad.net/ubuntu-push/bus/connectivity launchpad.net/ubuntu-push/bus/networkmanager launchpad.net/ubuntu-push/bus/notifications launchpad.net/ubuntu-push/bus/systemimage launchpad.net/ubuntu-push/bus/testing launchpad.net/ubuntu-push/bus/urldispatcher launchpad.net/ubuntu-push/client launchpad.net/ubuntu-push/client/gethosts launchpad.net/ubuntu-push/client/session launchpad.net/ubuntu-push/client/session/levelmap launchpad.net/ubuntu-push/config launchpad.net/ubuntu-push/external/murmur3 launchpad.net/ubuntu-push/logger launchpad.net/ubuntu-push/protocol launchpad.net/ubuntu-push/server launchpad.net/ubuntu-push/server/api launchpad.net/ubuntu-push/server/broker launchpad.net/ubuntu-push/server/broker/simple launchpad.net/ubuntu-push/server/broker/testing launchpad.net/ubuntu-push/server/broker/testsuite launchpad.net/ubuntu-push/server/dev launchpad.net/ubuntu-push/server/listener launchpad.net/ubuntu-push/server/session launchpad.net/ubuntu-push/server/store launchpad.net/ubuntu-push/testing launchpad.net/ubuntu-push/testing/condition launchpad.net/ubuntu-push/util launchpad.net/ubuntu-push/whoopsie launchpad.net/ubuntu-push/whoopsie/identifier launchpad.net/ubuntu-push/whoopsie/identifier/testing
? launchpad.net/ubuntu-push [no test files]
ok launchpad.net/ubuntu-push/bus 0.009s
ok launchpad.net/ubuntu-push/bus/connectivity 1.153s
ok launchpad.net/ubuntu-push/bus/networkmanager 0.099s
ok launchpad.net/ubuntu-push/bus/notifications 0.016s
ok launchpad.net/ubuntu-push/bus/systemimage 0.006s
ok launchpad.net/ubuntu-push/bus/testing 0.017s
ok launchpad.net/ubuntu-push/bus/urldispatcher 0.007s
ok launchpad.net/ubuntu-push/client 0.073s
ok launchpad.net/ubuntu-push/client/gethosts 0.710s
ok launchpad.net/ubuntu-push/client/session 0.130s
ok launchpad.net/ubuntu-push/client/session/levelmap 0.095s
ok launchpad.net/ubuntu-push/config 0.014s
ok launchpad.net/ubuntu-push/external/murmur3 0.011s
ok launchpad.net/ubuntu-push/logger 0.007s
ok launchpad.net/ubuntu-push/protocol 0.010s
ok launchpad.net/ubuntu-push/server 0.043s
ok launchpad.net/ubuntu-push/server/api 0.016s
ok launchpad.net/ubuntu-push/server/broker 0.009s
ok launchpad.net/ubuntu-push/server/broker/simple 0.006s
? launchpad.net/ubuntu-push/server/broker/testing [no test files]
? launchpad.net/ubuntu-push/server/broker/testsuite [no test files]
? launc...

=== modified file 'server/acceptance/acceptanceclient.go'
--- server/acceptance/acceptanceclient.go	2014-02-21 16:17:28 +0000
+++ server/acceptance/acceptanceclient.go	2014-04-04 09:57:13 +0000
@@ -35,6 +35,8 @@
 type ClientSession struct {
 	// configuration
 	DeviceId        string
+	Model           string
+	ImageChannel    string
 	ServerAddr      string
 	ExchangeTimeout time.Duration
 	CertPEMBlock    []byte
@@ -86,6 +88,10 @@
 		Type:     "connect",
 		DeviceId: sess.DeviceId,
 		Levels:   sess.Levels,
+		Info: map[string]interface{}{
+			"device":  sess.Model,
+			"channel": sess.ImageChannel,
+		},
 	})
 	if err != nil {
 		return err

=== modified file 'server/acceptance/cmd/acceptanceclient.go'
--- server/acceptance/cmd/acceptanceclient.go	2014-02-21 16:17:28 +0000
+++ server/acceptance/cmd/acceptanceclient.go	2014-04-04 09:57:13 +0000
@@ -30,6 +30,8 @@
 var (
 	insecureFlag    = flag.Bool("insecure", false, "disable checking of server certificate and hostname")
 	reportPingsFlag = flag.Bool("reportPings", true, "report each Ping from the server")
+	deviceModel     = flag.String("model", "?", "device image model")
+	imageChannel    = flag.String("imageChannel", "?", "image channel")
 )
 
 type configuration struct {
@@ -64,9 +66,12 @@
 		ServerAddr:      cfg.Addr.HostPort(),
 		DeviceId:        flag.Arg(1),
 		// flags
-		ReportPings: *reportPingsFlag,
-		Insecure:    *insecureFlag,
+		Model:        *deviceModel,
+		ImageChannel: *imageChannel,
+		ReportPings:  *reportPingsFlag,
+		Insecure:     *insecureFlag,
 	}
+	log.Printf("with: %#v", session)
 	session.CertPEMBlock, err = config.LoadFile(cfg.CertPEMFile, filepath.Dir(configFName))
 	if err != nil {
 		log.Fatalf("reading CertPEMFile: %v", err)

=== modified file 'server/acceptance/suites/broadcast.go'
--- server/acceptance/suites/broadcast.go	2014-02-21 21:39:54 +0000
+++ server/acceptance/suites/broadcast.go	2014-04-04 09:57:13 +0000
@@ -41,11 +41,34 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"n": 42}`),
-	})
-	c.Assert(err, IsNil)
-	c.Assert(got, Matches, ".*ok.*")
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
+		Data:     json.RawMessage(`{"img1/m1": 42}`),
+	})
+	c.Assert(err, IsNil)
+	c.Assert(got, Matches, ".*ok.*")
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"img1/m1":42}]`)
+	stop()
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *BroadcastAcceptanceSuite) TestBroadcastToConnectedChannelFilter(c *C) {
+	events, errCh, stop := s.StartClient(c, "DEVB", nil)
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"img1/m2": 10}`),
+	})
+	c.Assert(err, IsNil)
+	got, err = s.PostRequest("/broadcast", &api.Broadcast{
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"img1/m1": 20}`),
+	})
+	c.Assert(err, IsNil)
+	c.Assert(got, Matches, ".*ok.*")
+	// xxx don't send this one
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":20}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -56,14 +79,14 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 1}`),
+		Data:     json.RawMessage(`{"img1/m1": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 
 	events, errCh, stop := s.StartClient(c, "DEVB", nil)
 	// gettting pending on connect
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"b":1}]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"img1/m1":1}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -71,7 +94,7 @@
 
 func (s *BroadcastAcceptanceSuite) TestBroadcasLargeNeedsSplitting(c *C) {
 	// send bunch of broadcasts that will be pending
-	payloadFmt := fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
+	payloadFmt := fmt.Sprintf(`{"img1/m1":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
 	for i := 0; i < 32; i++ {
 		got, err := s.PostRequest("/broadcast", &api.Broadcast{
 			Channel:  "system",
@@ -84,7 +107,7 @@
 
 	events, errCh, stop := s.StartClient(c, "DEVC", nil)
 	// gettting pending on connect
-	c.Check(NextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:30 payloads:\[{"b":0,.*`)
+	c.Check(NextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:30 payloads:\[{"img1/m1":0,.*`)
 	c.Check(NextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:32 payloads:\[.*`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
@@ -100,12 +123,12 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"n": 42}`),
+		Data:     json.RawMessage(`{"img1/m1": 42}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	c.Check(NextEvent(events1, errCh1), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
-	c.Check(NextEvent(events2, errCh2), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
+	c.Check(NextEvent(events1, errCh1), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"img1/m1":42}]`)
+	c.Check(NextEvent(events2, errCh2), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"img1/m1":42}]`)
 	stop1()
 	stop2()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
@@ -119,11 +142,11 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 1}`),
+		Data:     json.RawMessage(`{"img1/m1": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"b":1}]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"img1/m1":1}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -131,7 +154,7 @@
 	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 2}`),
+		Data:     json.RawMessage(`{"img1/m1": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -139,7 +162,7 @@
 	events, errCh, stop = s.StartClient(c, "DEVD", map[string]int64{
 		protocol.SystemChannelId: 1,
 	})
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":2}]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":2}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -150,14 +173,14 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 1}`),
+		Data:     json.RawMessage(`{"img1/m1": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 2}`),
+		Data:     json.RawMessage(`{"img1/m1": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -166,7 +189,7 @@
 		protocol.SystemChannelId: 10,
 	})
 	// gettting last one pending on connect
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":2}]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":2}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -189,14 +212,14 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 1}`),
+		Data:     json.RawMessage(`{"img1/m1": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 2}`),
+		Data:     json.RawMessage(`{"img1/m1": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -205,7 +228,7 @@
 		protocol.SystemChannelId: -10,
 	})
 	// gettting pending on connect
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":1},{"b":2}]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":1},{"img1/m1":2}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -216,14 +239,14 @@
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
-		Data:     json.RawMessage(`{"b": 1}`),
+		Data:     json.RawMessage(`{"img1/m1": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: time.Now().Add(1 * time.Second).Format(time.RFC3339),
-		Data:     json.RawMessage(`{"b": 2}`),
+		Data:     json.RawMessage(`{"img1/m1": 2}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
@@ -233,7 +256,7 @@
 
 	events, errCh, stop := s.StartClient(c, "DEVB", nil)
 	// gettting pending on connect
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":1}]`)
+	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":1}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)

=== modified file 'server/acceptance/suites/pingpong.go'
--- server/acceptance/suites/pingpong.go	2014-02-21 20:29:16 +0000
+++ server/acceptance/suites/pingpong.go	2014-04-04 09:57:13 +0000
@@ -34,7 +34,7 @@
 func (s *PingPongAcceptanceSuite) TestConnectPingPing(c *C) {
 	errCh := make(chan error, 1)
 	events := make(chan string, 10)
-	sess := testClientSession(s.ServerAddr, "DEVA", true)
+	sess := testClientSession(s.ServerAddr, "DEVA", "m1", "img1", true)
 	err := sess.Dial()
 	c.Assert(err, IsNil)
 	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
@@ -68,7 +68,7 @@
 func (s *PingPongAcceptanceSuite) TestConnectPingNeverPong(c *C) {
 	errCh := make(chan error, 1)
 	events := make(chan string, 10)
-	sess := testClientSession(s.ServerAddr, "DEVB", true)
+	sess := testClientSession(s.ServerAddr, "DEVB", "m1", "img1", true)
 	err := sess.Dial()
 	c.Assert(err, IsNil)
 	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {

=== modified file 'server/acceptance/suites/suite.go'
--- server/acceptance/suites/suite.go	2014-03-31 16:43:15 +0000
+++ server/acceptance/suites/suite.go	2014-04-04 09:57:13 +0000
@@ -46,7 +46,7 @@
 func (h *ServerHandle) StartClient(c *C, devId string, levels map[string]int64) (events <-chan string, errorCh <-chan error, stop func()) {
 	errCh := make(chan error, 1)
 	cliEvents := make(chan string, 10)
-	sess := testClientSession(h.ServerAddr, devId, false)
+	sess := testClientSession(h.ServerAddr, devId, "m1", "img1", false)
 	sess.Levels = levels
 	err := sess.Dial()
 	c.Assert(err, IsNil)
@@ -127,7 +127,7 @@
 	return string(body), err
 }
 
-func testClientSession(addr string, deviceId string, reportPings bool) *acceptance.ClientSession {
+func testClientSession(addr string, deviceId, model, imageChannel string, reportPings bool) *acceptance.ClientSession {
 	certPEMBlock, err := ioutil.ReadFile(helpers.SourceRelative("../ssl/testing.cert"))
 	if err != nil {
 		panic(fmt.Sprintf("could not read ssl/testing.cert: %v", err))
@@ -137,6 +137,8 @@
 		ServerAddr:      addr,
 		CertPEMBlock:    certPEMBlock,
 		DeviceId:        deviceId,
+		Model:           model,
+		ImageChannel:    imageChannel,
 		ReportPings:     reportPings,
 	}
 }

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-02-21 16:04:44 +0000
+++ server/broker/broker.go	2014-04-04 09:57:13 +0000
@@ -49,6 +49,19 @@
 // LevelsMap is the type for holding channel levels for session.
 type LevelsMap map[store.InternalChannelId]int64
 
+// GetInfoString helps retrivieng a string out of a protocol.ConnectMsg.Info
+func GetInfoString(msg *protocol.ConnectMsg, name, defaultVal string) (string, error) {
+	v, ok := msg.Info[name]
+	if !ok {
+		return defaultVal, nil
+	}
+	s, ok := v.(string)
+	if !ok {
+		return "", ErrUnexpectedValue
+	}
+	return s, nil
+}
+
 // BrokerSession holds broker session state.
 type BrokerSession interface {
 	// SessionChannel returns the session control channel
@@ -56,6 +69,10 @@
 	SessionChannel() <-chan Exchange
 	// DeviceIdentifier returns the device id string.
 	DeviceIdentifier() string
+	// DeviceImageModel returns the device model.
+	DeviceImageModel() string
+	// DeviceImageChannel returns the device system image channel.
+	DeviceImageChannel() string
 	// Levels returns the current channel levels for the session
 	Levels() LevelsMap
 	// ExchangeScratchArea returns the scratch area for exchanges.
@@ -71,6 +88,9 @@
 	return fmt.Sprintf("session aborted (%s)", ea.Reason)
 }
 
+// Unexpect value in message
+var ErrUnexpectedValue = &ErrAbort{"unexpected value in message"}
+
 // BrokerConfig gives access to the typical broker configuration.
 type BrokerConfig interface {
 	// SessionQueueSize gives the session queue size.

=== modified file 'server/broker/broker_test.go'
--- server/broker/broker_test.go	2014-02-10 23:19:08 +0000
+++ server/broker/broker_test.go	2014-04-04 09:57:13 +0000
@@ -20,6 +20,8 @@
 	"fmt"
 
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/protocol"
 )
 
 type brokerSuite struct{}
@@ -30,3 +32,19 @@
 	err := &ErrAbort{"expected FOO"}
 	c.Check(fmt.Sprintf("%s", err), Equals, "session aborted (expected FOO)")
 }
+
+func (s *brokerSuite) TestGetInfoString(c *C) {
+	connectMsg := &protocol.ConnectMsg{}
+	v, err := GetInfoString(connectMsg, "foo", "?")
+	c.Check(err, IsNil)
+	c.Check(v, Equals, "?")
+
+	connectMsg.Info = map[string]interface{}{"foo": "yay"}
+	v, err = GetInfoString(connectMsg, "foo", "?")
+	c.Check(err, IsNil)
+	c.Check(v, Equals, "yay")
+
+	connectMsg.Info["foo"] = 33
+	v, err = GetInfoString(connectMsg, "foo", "?")
+	c.Check(err, Equals, ErrUnexpectedValue)
+}

=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-02-26 16:04:57 +0000
+++ server/broker/exchanges.go	2014-04-04 09:57:13 +0000
@@ -18,6 +18,7 @@
 
 import (
 	"encoding/json"
+	"fmt"
 
 	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/store"
@@ -37,11 +38,24 @@
 	ChanId               store.InternalChannelId
 	TopLevel             int64
 	NotificationPayloads []json.RawMessage
+	Decoded              []map[string]interface{}
 }
 
 // check interface already here
 var _ Exchange = &BroadcastExchange{}
 
+// Init ensures the BroadcastExchange is fully initialized for the sessions.
+func (sbe *BroadcastExchange) Init() {
+	decoded := make([]map[string]interface{}, len(sbe.NotificationPayloads))
+	sbe.Decoded = decoded
+	for i, p := range sbe.NotificationPayloads {
+		err := json.Unmarshal(p, &decoded[i])
+		if err != nil {
+			decoded[i] = nil
+		}
+	}
+}
+
 func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
 	c := int64(len(payloads))
 	if c == 0 {
@@ -58,6 +72,20 @@
 	}
 }
 
+func channelFilter(tag string, chanId store.InternalChannelId, payloads []json.RawMessage, decoded []map[string]interface{}) []json.RawMessage {
+	if len(payloads) != 0 && chanId == store.SystemInternalChannelId {
+		decoded := decoded[len(decoded)-len(payloads):]
+		filtered := make([]json.RawMessage, 0)
+		for i, decoded1 := range decoded {
+			if _, ok := decoded1[tag]; ok {
+				filtered = append(filtered, payloads[i])
+			}
+		}
+		payloads = filtered
+	}
+	return payloads
+}
+
 // Prepare session for a BROADCAST.
 func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
 	scratchArea := sess.ExchangeScratchArea()
@@ -65,6 +93,9 @@
 	scratchArea.broadcastMsg.Type = "broadcast"
 	clientLevel := sess.Levels()[sbe.ChanId]
 	payloads := filterByLevel(clientLevel, sbe.TopLevel, sbe.NotificationPayloads)
+	tag := fmt.Sprintf("%s/%s", sess.DeviceImageChannel(), sess.DeviceImageModel())
+	payloads = channelFilter(tag, sbe.ChanId, payloads, sbe.Decoded)
+
 	// xxx need an AppId as well, later
 	scratchArea.broadcastMsg.ChanId = store.InternalChannelIdToHex(sbe.ChanId)
 	scratchArea.broadcastMsg.TopLevel = sbe.TopLevel

=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-02-26 16:04:57 +0000
+++ server/broker/exchanges_test.go	2014-04-04 09:57:13 +0000
@@ -35,24 +35,45 @@
 
 var _ = Suite(&exchangesSuite{})
 
+func (s *exchangesSuite) TestBroadcastExchangeInit(c *C) {
+	exchg := &broker.BroadcastExchange{
+		ChanId:   store.SystemInternalChannelId,
+		TopLevel: 3,
+		NotificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"x"}`),
+			json.RawMessage(`[]`),
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	exchg.Init()
+	c.Check(exchg.Decoded, DeepEquals, []map[string]interface{}{
+		map[string]interface{}{"a": "x"},
+		nil,
+		map[string]interface{}{"a": "y"},
+	})
+}
+
 func (s *exchangesSuite) TestBroadcastExchange(c *C) {
 	sess := &testing.TestBrokerSession{
-		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		Model:        "m1",
+		ImageChannel: "img1",
 	}
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
 		NotificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"x"}`),
-			json.RawMessage(`{"a":"y"}`),
+			json.RawMessage(`{"img1/m1":100}`),
+			json.RawMessage(`{"img2/m2":200}`),
 		},
 	}
+	exchg.Init()
 	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":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"x"},{"a":"y"}]}`)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"img1/m1":100}]}`)
 	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
 	c.Assert(err, IsNil)
 	err = exchg.Acked(sess, true)
@@ -62,9 +83,11 @@
 
 func (s *exchangesSuite) TestBroadcastExchangeReuseVsSplit(c *C) {
 	sess := &testing.TestBrokerSession{
-		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		Model:        "m1",
+		ImageChannel: "img1",
 	}
-	payloadFmt := fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
+	payloadFmt := fmt.Sprintf(`{"img1/m1":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
 	needsSplitting := make([]json.RawMessage, 32)
 	for i := 0; i < 32; i++ {
 		needsSplitting[i] = json.RawMessage(fmt.Sprintf(payloadFmt, i))
@@ -76,6 +99,7 @@
 		TopLevel:             topLevel,
 		NotificationPayloads: needsSplitting,
 	}
+	exchg.Init()
 	outMsg, _, err := exchg.Prepare(sess)
 	c.Assert(err, IsNil)
 	parts := 0
@@ -91,10 +115,11 @@
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: topLevel + 2,
 		NotificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"x"}`),
-			json.RawMessage(`{"a":"y"}`),
+			json.RawMessage(`{"img1/m1":"x"}`),
+			json.RawMessage(`{"img1/m1":"y"}`),
 		},
 	}
+	exchg.Init()
 	outMsg, _, err = exchg.Prepare(sess)
 	c.Assert(err, IsNil)
 	done := outMsg.Split() // shouldn't panic
@@ -103,21 +128,24 @@
 
 func (s *exchangesSuite) TestBroadcastExchangeAckMismatch(c *C) {
 	sess := &testing.TestBrokerSession{
-		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		Model:        "m1",
+		ImageChannel: "img2",
 	}
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
 		NotificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"y"}`),
+			json.RawMessage(`{"img2/m1":1}`),
 		},
 	}
+	exchg.Init()
 	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":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"img2/m1":1}]}`)
 	err = json.Unmarshal([]byte(`{}`), inMsg)
 	c.Assert(err, IsNil)
 	err = exchg.Acked(sess, true)
@@ -130,23 +158,55 @@
 		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{
 			store.SystemInternalChannelId: 2,
 		}),
+		Model:        "m1",
+		ImageChannel: "img1",
 	}
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
 		NotificationPayloads: []json.RawMessage{
-			json.RawMessage(`{"a":"x"}`),
-			json.RawMessage(`{"a":"y"}`),
-		},
-	}
-	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":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
-	c.Assert(err, IsNil)
-	err = exchg.Acked(sess, true)
-	c.Assert(err, IsNil)
+			json.RawMessage(`{"img1/m1":100}`),
+			json.RawMessage(`{"img1/m1":101}`),
+		},
+	}
+	exchg.Init()
+	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":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"img1/m1":101}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, IsNil)
+}
+
+func (s *exchangesSuite) TestBroadcastExchangeChannelFilter(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		Model:        "m1",
+		ImageChannel: "img1",
+	}
+	exchg := &broker.BroadcastExchange{
+		ChanId:   store.SystemInternalChannelId,
+		TopLevel: 5,
+		NotificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"img1/m1":100}`),
+			json.RawMessage(`{"img2/m2":200}`),
+			json.RawMessage(`{"img1/m1":101}`),
+		},
+	}
+	exchg.Init()
+	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":"broadcast","ChanId":"0","TopLevel":5,"Payloads":[{"img1/m1":100},{"img1/m1":101}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, IsNil)
+	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(5))
 }

=== modified file 'server/broker/exchg_impl_test.go'
--- server/broker/exchg_impl_test.go	2014-02-10 23:19:08 +0000
+++ server/broker/exchg_impl_test.go	2014-04-04 09:57:13 +0000
@@ -20,6 +20,8 @@
 	"encoding/json"
 
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/server/store"
 )
 
 type exchangesImplSuite struct{}
@@ -56,3 +58,31 @@
 	res = filterByLevel(5, 10, nil)
 	c.Check(len(res), Equals, 0)
 }
+
+func (s *exchangesImplSuite) TestChannelFilter(c *C) {
+	payloads := []json.RawMessage{
+		json.RawMessage(`{"a/x": 3}`),
+		json.RawMessage(`{"b/x": 4}`),
+		json.RawMessage(`{"a/y": 5}`),
+		json.RawMessage(`{"a/x": 6}`),
+	}
+	decoded := make([]map[string]interface{}, 4)
+	for i, p := range payloads {
+		err := json.Unmarshal(p, &decoded[i])
+		c.Assert(err, IsNil)
+	}
+
+	other := store.InternalChannelId("1")
+
+	c.Check(channelFilter("", store.SystemInternalChannelId, nil, nil), IsNil)
+	c.Check(channelFilter("", other, payloads[1:], decoded), DeepEquals, payloads[1:])
+
+	// use tag when channel is the sytem channel
+
+	c.Check(channelFilter("c/z", store.SystemInternalChannelId, payloads, decoded), HasLen, 0)
+
+	c.Check(channelFilter("a/x", store.SystemInternalChannelId, payloads, decoded), DeepEquals, []json.RawMessage{payloads[0], payloads[3]})
+
+	c.Check(channelFilter("a/x", store.SystemInternalChannelId, payloads[1:], decoded), DeepEquals, []json.RawMessage{payloads[3]})
+
+}

=== modified file 'server/broker/simple/simple.go'
--- server/broker/simple/simple.go	2014-02-21 16:04:44 +0000
+++ server/broker/simple/simple.go	2014-04-04 09:57:13 +0000
@@ -46,11 +46,13 @@
 
 // simpleBrokerSession represents a session in the broker.
 type simpleBrokerSession struct {
-	registered bool
-	deviceId   string
-	done       chan bool
-	exchanges  chan broker.Exchange
-	levels     broker.LevelsMap
+	registered   bool
+	deviceId     string
+	model        string
+	imageChannel string
+	done         chan bool
+	exchanges    chan broker.Exchange
+	levels       broker.LevelsMap
 	// for exchanges
 	exchgScratch broker.ExchangesScratchArea
 }
@@ -75,6 +77,14 @@
 	return sess.deviceId
 }
 
+func (sess *simpleBrokerSession) DeviceImageModel() string {
+	return sess.model
+}
+
+func (sess *simpleBrokerSession) DeviceImageChannel() string {
+	return sess.imageChannel
+}
+
 func (sess *simpleBrokerSession) Levels() broker.LevelsMap {
 	return sess.levels
 }
@@ -147,6 +157,7 @@
 				TopLevel:             topLevel,
 				NotificationPayloads: payloads,
 			}
+			broadcastExchg.Init()
 			sess.exchanges <- broadcastExchg
 		}
 	}
@@ -157,6 +168,14 @@
 // pending notifications as well.
 func (b *SimpleBroker) Register(connect *protocol.ConnectMsg) (broker.BrokerSession, error) {
 	// xxx sanity check DeviceId
+	model, err := broker.GetInfoString(connect, "device", "?")
+	if err != nil {
+		return nil, err
+	}
+	imageChannel, err := broker.GetInfoString(connect, "channel", "?")
+	if err != nil {
+		return nil, err
+	}
 	levels := map[store.InternalChannelId]int64{}
 	for hexId, v := range connect.Levels {
 		id, err := store.HexToInternalChannelId(hexId)
@@ -166,14 +185,16 @@
 		levels[id] = v
 	}
 	sess := &simpleBrokerSession{
-		deviceId:  connect.DeviceId,
-		done:      make(chan bool),
-		exchanges: make(chan broker.Exchange, b.sessionQueueSize),
-		levels:    levels,
+		deviceId:     connect.DeviceId,
+		model:        model,
+		imageChannel: imageChannel,
+		done:         make(chan bool),
+		exchanges:    make(chan broker.Exchange, b.sessionQueueSize),
+		levels:       levels,
 	}
 	b.sessionCh <- sess
 	<-sess.done
-	err := b.feedPending(sess)
+	err = b.feedPending(sess)
 	if err != nil {
 		return nil, err
 	}
@@ -219,6 +240,7 @@
 					TopLevel:             topLevel,
 					NotificationPayloads: payloads,
 				}
+				broadcastExchg.Init()
 				for _, sess := range b.registry {
 					sess.exchanges <- broadcastExchg
 				}

=== modified file 'server/broker/simple/simple_test.go'
--- server/broker/simple/simple_test.go	2014-02-10 23:29:53 +0000
+++ server/broker/simple/simple_test.go	2014-04-04 09:57:13 +0000
@@ -48,6 +48,7 @@
 	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{
@@ -60,6 +61,7 @@
 		ChanId:               store.SystemInternalChannelId,
 		TopLevel:             1,
 		NotificationPayloads: []json.RawMessage{notification1},
+		Decoded:              []map[string]interface{}{decoded1},
 	})
 }
 

=== modified file 'server/broker/testing/impls.go'
--- server/broker/testing/impls.go	2014-01-23 20:13:22 +0000
+++ server/broker/testing/impls.go	2014-04-04 09:57:13 +0000
@@ -24,6 +24,8 @@
 // Test implementation of BrokerSession.
 type TestBrokerSession struct {
 	DeviceId     string
+	Model        string
+	ImageChannel string
 	Exchanges    chan broker.Exchange
 	LevelsMap    broker.LevelsMap
 	exchgScratch broker.ExchangesScratchArea
@@ -33,6 +35,14 @@
 	return tbs.DeviceId
 }
 
+func (tbs *TestBrokerSession) DeviceImageModel() string {
+	return tbs.Model
+}
+
+func (tbs *TestBrokerSession) DeviceImageChannel() string {
+	return tbs.ImageChannel
+}
+
 func (tbs *TestBrokerSession) SessionChannel() <-chan broker.Exchange {
 	return tbs.Exchanges
 }

=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-03-19 23:46:18 +0000
+++ server/broker/testsuite/suite.go	2014-04-04 09:57:13 +0000
@@ -81,10 +81,20 @@
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
-	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1", Levels: map[string]int64{"0": 5}})
+	sess, err := b.Register(&protocol.ConnectMsg{
+		Type:     "connect",
+		DeviceId: "dev-1",
+		Levels:   map[string]int64{"0": 5},
+		Info: map[string]interface{}{
+			"device":  "model",
+			"channel": "daily",
+		},
+	})
 	c.Assert(err, IsNil)
 	c.Assert(s.RevealSession(b, "dev-1"), Equals, sess)
 	c.Assert(sess.DeviceIdentifier(), Equals, "dev-1")
+	c.Check(sess.DeviceImageModel(), Equals, "model")
+	c.Check(sess.DeviceImageChannel(), Equals, "daily")
 	c.Assert(sess.ExchangeScratchArea(), Not(IsNil))
 	c.Check(sess.Levels(), DeepEquals, broker.LevelsMap(map[store.InternalChannelId]int64{
 		store.SystemInternalChannelId: 5,
@@ -105,6 +115,22 @@
 	c.Check(err, FitsTypeOf, &broker.ErrAbort{})
 }
 
+func (s *CommonBrokerSuite) TestRegistrationInfoErrors(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	b := s.MakeBroker(sto, testBrokerConfig, nil)
+	b.Start()
+	defer b.Stop()
+	info := map[string]interface{}{
+		"device": -1,
+	}
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", Info: info})
+	c.Check(err, Equals, broker.ErrUnexpectedValue)
+	info["device"] = "m"
+	info["channel"] = -1
+	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", Info: info})
+	c.Check(err, Equals, broker.ErrUnexpectedValue)
+}
+
 func (s *CommonBrokerSuite) TestRegistrationFeedPending(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	notification1 := json.RawMessage(`{"m": "M"}`)
@@ -149,6 +175,7 @@
 func (s *CommonBrokerSuite) TestBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	notification1 := json.RawMessage(`{"m": "M"}`)
+	decoded1 := map[string]interface{}{"m": "M"}
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
@@ -168,6 +195,7 @@
 			ChanId:               store.SystemInternalChannelId,
 			TopLevel:             1,
 			NotificationPayloads: []json.RawMessage{notification1},
+			Decoded:              []map[string]interface{}{decoded1},
 		})
 	}
 	select {
@@ -178,6 +206,7 @@
 			ChanId:               store.SystemInternalChannelId,
 			TopLevel:             1,
 			NotificationPayloads: []json.RawMessage{notification1},
+			Decoded:              []map[string]interface{}{decoded1},
 		})
 	}
 }