Merge lp:~pedronis/ubuntu-push/register-for-unicast into lp:ubuntu-push

[Samuele Pedroni]

support registering tokens and sending notifications with a token; register script and scripts unicast support

support registering tokens and sending notifications with a token

register script and scripts unicast support

other small tweaks

[lp:~pedronis/ubuntu-push/register-for-unicast]

182. By Samuele Pedroni on 2014-06-05

move register to py3

Unmerged revisions

=== modified file '.bzrignore'
--- .bzrignore	2014-03-31 16:43:15 +0000
+++ .bzrignore	2014-06-05 12:03:59 +0000
@@ -13,3 +13,12 @@
 debian/*.substvars
 ubuntu-push-client
 push-server-dev
+signing-helper/CMakeCache.txt
+signing-helper/CMakeFiles
+signing-helper/Makefile
+signing-helper/cmake_install.cmake
+signing-helper/moc_signing.cpp
+signing-helper/signing-helper
+signing-helper/signing-helper_automoc.cpp
+.has-fetched-deps
+*.deps

=== modified file 'Makefile'
--- Makefile	2014-03-31 17:58:54 +0000
+++ Makefile	2014-06-05 12:03:59 +0000
@@ -11,10 +11,24 @@
 GODEPS += launchpad.net/go-dbus/v1
 GODEPS += launchpad.net/go-xdg/v0
 GODEPS += code.google.com/p/gosqlite/sqlite3
+GODEPS += code.google.com/p/go-uuid/uuid
 
 TOTEST = $(shell env GOPATH=$(GOPATH) go list $(PROJECT)/...|grep -v acceptance|grep -v http13client )
-
-bootstrap:
+TOBUILD = $(shell grep -lr '^package main')
+
+all: fetchdeps bootstrap build-client build-server-dev
+
+fetchdeps: .has-fetched-deps
+
+.has-fetched-deps: PACKAGE_DEPS
+	@$(MAKE) --no-print-directory refetchdeps
+	@touch $@
+
+refetchdeps:
+	sudo apt-get install $$( grep -v '^#' PACKAGE_DEPS )
+
+bootstrap: dependencies.tsv
+	$(RM) -r $(GOPATH)/pkg
 	mkdir -p $(GOPATH)/bin
 	mkdir -p $(GOPATH)/pkg
 	go get -u launchpad.net/godeps
@@ -22,6 +36,9 @@
 	$(GOPATH)/bin/godeps -u dependencies.tsv
 	go install $(GODEPS)
 
+dependencies.tsv: $(TOBUILD)
+	$(GOPATH)/bin/godeps -t $(foreach i,$^,$(dir $(PROJECT)/$(i))) 2>/dev/null | cat > $@
+
 check:
 	go test $(TESTFLAGS) $(TOTEST)
 
@@ -31,14 +48,39 @@
 acceptance:
 	cd server/acceptance; ./acceptance.sh
 
-build-client:
-	go build ubuntu-push-client.go
-
-build-server-dev:
-	go build -o push-server-dev launchpad.net/ubuntu-push/server/dev
-
-run-server-dev:
-	go run server/dev/*.go sampleconfigs/dev.json
+build-client: ubuntu-push-client signing-helper/signing-helper
+
+%.deps: %
+	$(SH) scripts/deps.sh $<
+
+%: %.go
+	go build -o $@ $<
+
+include $(TOBUILD:.go=.go.deps)
+
+signing-helper/Makefile: signing-helper/CMakeLists.txt signing-helper/signing-helper.cpp signing-helper/signing.h
+	cd signing-helper && (make clean || true) && cmake .
+
+signing-helper/signing-helper: signing-helper/Makefile signing-helper/signing-helper.cpp signing-helper/signing.h
+	cd signing-helper && make
+
+build-server-dev: push-server-dev
+
+run-server-dev: push-server-dev
+	./$< sampleconfigs/dev.json
+
+push-server-dev: server/dev/server
+	mv $< $@
+
+# very basic cleanup stuff; needs more work
+clean:
+	$(RM) -r coverhtml
+	$(MAKE) -C signing-helper clean || true
+	$(RM) push-server-dev
+	$(RM) $(TOBUILD:.go=)
+
+distclean:
+	bzr clean-tree --verbose --ignored --force
 
 coverage-summary:
 	go test $(TESTFLAGS) -a -cover $(TOTEST)
@@ -67,5 +109,8 @@
 	dot -Tsvg $< > $@
 
 .PHONY: bootstrap check check-race format check-format \
-	acceptance build-client build server-dev run-server-dev \
-	coverage-summary coverage-html protocol-diagrams
+	acceptance build-client build-server-dev run-server-dev \
+	coverage-summary coverage-html protocol-diagrams \
+	fetchdeps refetchdeps clean distclean all
+
+.INTERMEDIATE: server/dev/server

=== added file 'PACKAGE_DEPS'
--- PACKAGE_DEPS	1970-01-01 00:00:00 +0000
+++ PACKAGE_DEPS	2014-06-05 12:03:59 +0000
@@ -0,0 +1,10 @@
+# See the README for what this file is and how to use it.
+build-essential
+cmake
+libdbus-1-dev
+libgcrypt11-dev
+libglib2.0-dev
+libnih-dbus-dev
+libsqlite3-dev
+libubuntuoneauth-2.0-dev
+libwhoopsie-dev

=== modified file 'README'
--- README	2014-03-31 17:58:54 +0000
+++ README	2014-06-05 12:03:59 +0000
@@ -6,11 +6,23 @@
 The code expects to be checked out as launchpad.net/ubuntu-push in a Go
 workspace, see "go help gopath".
 
-To setup Go dependencies, install libsqlite3-dev and run:
-
-   make bootstrap
-
-To run tests, install libgcrypt11-dev and libwhoopsie-dev and run:
+You need a somewhat long list of dependencies, as well as a working Go
+development environment. THe Ubuntu packagenames for these are listed
+in the file PACKAGE_DEPS.
+
+On Ubuntu, if you have sudo, you can have all those installed for you
+by do doing
+
+    make fetchdeps
+
+Once you have the packaged dependencies you can get the Go
+dependencies via
+
+    make bootstrap
+
+and then you're set. Good luck!
+
+To run the tests:
 
    make check
 

=== modified file 'bus/connectivity/connectivity.go'
--- bus/connectivity/connectivity.go	2014-04-04 11:08:28 +0000
+++ bus/connectivity/connectivity.go	2014-06-05 12:03:59 +0000
@@ -72,19 +72,20 @@
 		cs.connAttempts += ar.Redial()
 		nm := networkmanager.New(cs.endp, cs.log)
 
+		// set up the watch
+		stateCh, err = nm.WatchState()
+		if err != nil {
+			cs.log.Debugf("failed to set up the state watch: %s", err)
+			goto Continue
+		}
+
 		// Get the current state.
 		initial = nm.GetState()
 		if initial == networkmanager.Unknown {
 			cs.log.Debugf("Failed to get state.")
 			goto Continue
 		}
-
-		// set up the watch
-		stateCh, err = nm.WatchState()
-		if err != nil {
-			cs.log.Debugf("failed to set up the state watch: %s", err)
-			goto Continue
-		}
+		cs.log.Debugf("got initial state of %s", initial)
 
 		primary = nm.GetPrimaryConnection()
 		cs.log.Debugf("primary connection starts as %#v", primary)

=== modified file 'bus/connectivity/connectivity_test.go'
--- bus/connectivity/connectivity_test.go	2014-04-04 12:01:42 +0000
+++ bus/connectivity/connectivity_test.go	2014-06-05 12:03:59 +0000
@@ -17,8 +17,15 @@
 package connectivity
 
 import (
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
 	"launchpad.net/go-dbus/v1"
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/bus"
 	"launchpad.net/ubuntu-push/bus/networkmanager"
 	testingbus "launchpad.net/ubuntu-push/bus/testing"
 	"launchpad.net/ubuntu-push/config"
@@ -26,9 +33,6 @@
 	helpers "launchpad.net/ubuntu-push/testing"
 	"launchpad.net/ubuntu-push/testing/condition"
 	"launchpad.net/ubuntu-push/util"
-	"net/http/httptest"
-	"testing"
-	"time"
 )
 
 // hook up gocheck
@@ -115,6 +119,79 @@
 	c.Check(<-cs.networkStateCh, Equals, networkmanager.ConnectedGlobal)
 }
 
+// a racyEndpoint is an endpoint that behaves differently depending on
+// how much time passes between getting the state and setting up the
+// watch
+type racyEndpoint struct {
+	stateGot bool
+	maxTime  time.Time
+	delta    time.Duration
+	lock     sync.RWMutex
+}
+
+func (rep *racyEndpoint) GetProperty(prop string) (interface{}, error) {
+	switch prop {
+	case "state":
+		rep.lock.Lock()
+		defer rep.lock.Unlock()
+		rep.stateGot = true
+		rep.maxTime = time.Now().Add(rep.delta)
+		return uint32(networkmanager.Connecting), nil
+	case "PrimaryConnection":
+		return dbus.ObjectPath("/something"), nil
+	default:
+		return nil, nil
+	}
+}
+
+func (rep *racyEndpoint) WatchSignal(member string, f func(...interface{}), d func()) error {
+	if member == "StateChanged" {
+		// we count never having gotten the state as happening "after" now.
+		rep.lock.RLock()
+		defer rep.lock.RUnlock()
+		ok := !rep.stateGot || time.Now().Before(rep.maxTime)
+		go func() {
+			if ok {
+				f(uint32(networkmanager.ConnectedGlobal))
+			}
+			d()
+		}()
+	}
+	return nil
+}
+
+func (*racyEndpoint) Close()                                           {}
+func (*racyEndpoint) Dial() error                                      { return nil }
+func (*racyEndpoint) String() string                                   { return "racyEndpoint" }
+func (*racyEndpoint) Call(string, []interface{}, ...interface{}) error { return nil }
+func (*racyEndpoint) GrabName(bool) <-chan error                       { return nil }
+func (*racyEndpoint) WatchMethod(bus.DispatchMap, ...interface{})      {}
+func (*racyEndpoint) Signal(member string, args []interface{}) error   { return nil }
+
+var _ bus.Endpoint = (*racyEndpoint)(nil)
+
+// takeNext takes a value from given channel with a 1s timeout
+func takeNext(ch <-chan networkmanager.State) networkmanager.State {
+	select {
+	case <-time.After(time.Second):
+		panic("channel stuck: too long waiting")
+	case v := <-ch:
+		return v
+	}
+}
+
+// test that if the nm state goes from connecting to connected very
+// shortly after calling GetState, we don't lose the event.
+func (s *ConnSuite) TestStartAvoidsRace(c *C) {
+	for delta := time.Second; delta > 1; delta /= 2 {
+		rep := &racyEndpoint{delta: delta}
+		cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: rep}
+		f := Commentf("when delta=%s", delta)
+		c.Assert(cs.start(), Equals, networkmanager.Connecting, f)
+		c.Assert(takeNext(cs.networkStateCh), Equals, networkmanager.ConnectedGlobal, f)
+	}
+}
+
 /*
    tests for connectedStateStep()
 */

=== modified file 'bus/endpoint.go'
--- bus/endpoint.go	2014-04-02 08:23:15 +0000
+++ bus/endpoint.go	2014-06-05 12:03:59 +0000
@@ -29,9 +29,15 @@
  *    Endpoint (and its implementation)
  */
 
+type BusMethod func([]interface{}, []interface{}) ([]interface{}, error)
+type DispatchMap map[string]BusMethod
+
 // bus.Endpoint represents the DBus connection itself.
 type Endpoint interface {
+	GrabName(allowReplacement bool) <-chan error
 	WatchSignal(member string, f func(...interface{}), d func()) error
+	WatchMethod(DispatchMap, ...interface{})
+	Signal(string, []interface{}) error
 	Call(member string, args []interface{}, rvs ...interface{}) error
 	GetProperty(property string) (interface{}, error)
 	Dial() error
@@ -53,13 +59,18 @@
 }
 
 // ensure endpoint implements Endpoint
-var _ Endpoint = &endpoint{}
+var _ Endpoint = (*endpoint)(nil)
 
 /*
    public methods
+
+XXX:   these are almost entirely untested, as that would need
+XXX:   integration tests we are currently missing.
 */
 
 // Dial() (re)establishes the connection with dbus
+//
+// XXX: mostly untested
 func (endp *endpoint) Dial() error {
 	bus, err := dbus.Connect(endp.busT.(concreteBus).dbusType())
 	if err != nil {
@@ -106,6 +117,8 @@
 // with the unpacked value. If it's unable to set up the watch it returns an
 // error. If the watch fails once established, d() is called. Typically f()
 // sends the values over a channel, and d() would close the channel.
+//
+// XXX: untested
 func (endp *endpoint) WatchSignal(member string, f func(...interface{}), d func()) error {
 	watch, err := endp.proxy.WatchSignal(endp.addr.Interface, member)
 	if err != nil {
@@ -122,6 +135,8 @@
 // interface provided when creating the endpoint). args can be built
 // using bus.Args(...). The return value is unpacked into rvs before being
 // returned.
+//
+// XXX: untested
 func (endp *endpoint) Call(member string, args []interface{}, rvs ...interface{}) error {
 	msg, err := endp.proxy.Call(endp.addr.Interface, member, args...)
 	if err != nil {
@@ -138,6 +153,8 @@
 // to read a given property on the name, path and interface provided when
 // creating the endpoint. The return value is unpacked into a dbus.Variant,
 // and its value returned.
+//
+// XXX: untested
 func (endp *endpoint) GetProperty(property string) (interface{}, error) {
 	msg, err := endp.proxy.Call("org.freedesktop.DBus.Properties", "Get", endp.addr.Interface, property)
 	if err != nil {
@@ -160,6 +177,8 @@
 }
 
 // Close the connection to dbus.
+//
+// XXX: untested
 func (endp *endpoint) Close() {
 	if endp.bus != nil {
 		endp.bus.Close()
@@ -169,15 +188,98 @@
 }
 
 // String() performs advanced endpoint stringification
+//
+// XXX: untested
 func (endp *endpoint) String() string {
 	return fmt.Sprintf("<Connection to %s %#v>", endp.bus, endp.addr)
 }
 
+// GrabName() takes over the name on the bus, reporting errors over the
+// returned channel.
+//
+// While the first result will be nil on success, successive results would
+// typically indicate another process trying to take over the name.
+//
+// XXX: untested
+func (endp *endpoint) GrabName(allowReplacement bool) <-chan error {
+	flags := dbus.NameFlagAllowReplacement | dbus.NameFlagReplaceExisting
+	if !allowReplacement {
+		flags = 0
+	}
+	return endp.bus.RequestName(endp.addr.Name, flags).C
+}
+
+// Signal() sends out a signal called <member> containing <args>.
+//
+// XXX: untested
+func (endp *endpoint) Signal(member string, args []interface{}) error {
+	msg := dbus.NewSignalMessage(dbus.ObjectPath(endp.addr.Path), endp.addr.Interface, member)
+	if args != nil {
+		err := msg.AppendArgs(args...)
+		if err != nil {
+			endp.log.Errorf("unable to build dbus signal message: %v", err)
+			return err
+		}
+	}
+	err := endp.bus.Send(msg)
+	if err != nil {
+		endp.log.Errorf("unable to send dbus signal: %v", err)
+	} else {
+		endp.log.Debugf("sent dbus signal %s(%#v)", member, args)
+	}
+	return nil
+}
+
+// WatchMethod() uses the given DispatchMap to answer incoming method
+// calls.
+//
+// XXX: untested
+func (endp *endpoint) WatchMethod(dispatch DispatchMap, extra ...interface{}) {
+	ch := make(chan *dbus.Message)
+	go func() {
+		var reply *dbus.Message
+
+		err_iface := endp.addr.Interface + ".Error"
+
+		for msg := range ch {
+			meth, ok := dispatch[msg.Member]
+			if !ok || msg.Interface != endp.addr.Interface {
+				reply = dbus.NewErrorMessage(msg,
+					"org.freedesktop.DBus.Error.UnknownMethod", "Unknown method")
+				endp.log.Errorf("WatchMethod: unknown method %s", msg.Member)
+			} else {
+				args := msg.AllArgs()
+				rvals, err := meth(args, extra)
+				if err != nil {
+					reply = dbus.NewErrorMessage(msg, err_iface, err.Error())
+					endp.log.Errorf("WatchMethod: %s(%#v, %#v) failure: %#v", msg.Member, args, extra, err)
+				} else {
+					endp.log.Debugf("WatchMethod: %s(%#v, %#v) success: %#v", msg.Member, args, extra, rvals)
+					reply = dbus.NewMethodReturnMessage(msg)
+					err = reply.AppendArgs(rvals...)
+					if err != nil {
+						endp.log.Errorf("WatchMethod: unable to build dbus response message: %v", err)
+						reply = dbus.NewErrorMessage(msg, err_iface, err.Error())
+					}
+				}
+			}
+			err := endp.bus.Send(reply)
+			if err != nil {
+				endp.log.Errorf("WatchMethod: unable to send reply: %v", err)
+			}
+
+		}
+	}()
+	endp.bus.RegisterObjectPath(dbus.ObjectPath(endp.addr.Path), ch)
+}
+
 /*
    private methods
 */
 
 // unpackOneMsg unpacks the value from the response msg
+//
+// XXX: untested
 func (endp *endpoint) unpackOneMsg(msg *dbus.Message, member string) []interface{} {
 	var varmap map[string]dbus.Variant
 	if err := msg.Args(&varmap); err != nil {
@@ -187,6 +289,8 @@
 }
 
 // unpackMessages unpacks the value from the watch
+//
+// XXX: untested
 func (endp *endpoint) unpackMessages(watch *dbus.SignalWatch, f func(...interface{}), d func(), member string) {
 	for {
 		msg, ok := <-watch.C

=== modified file 'bus/endpoint_test.go'
--- bus/endpoint_test.go	2014-02-06 09:57:49 +0000
+++ bus/endpoint_test.go	2014-06-05 12:03:59 +0000
@@ -37,7 +37,7 @@
 // testing amenities (already talked about it with jamesh)
 
 // Tests that we can connect to the *actual* system bus.
-// XXX maybe connect to a mock/fake/etc bus?
+// XXX: maybe connect to a mock/fake/etc bus?
 func (s *EndpointSuite) TestDial(c *C) {
 	// if somebody's set up the env var, assume it's "live"
 	if os.Getenv("DBUS_SYSTEM_BUS_ADDRESS") == "" {

=== modified file 'bus/notifications/raw.go'
--- bus/notifications/raw.go	2014-04-02 08:23:15 +0000
+++ bus/notifications/raw.go	2014-06-05 12:03:59 +0000
@@ -22,6 +22,8 @@
 // this is the lower-level api
 
 import (
+	"errors"
+
 	"launchpad.net/go-dbus/v1"
 	"launchpad.net/ubuntu-push/bus"
 	"launchpad.net/ubuntu-push/logger"
@@ -68,6 +70,9 @@
 	timeout int32) (uint32, error) {
 	// that's a long argument list! Take a breather.
 	//
+	if raw.bus == nil {
+		return 0, errors.New("unconfigured (missing bus)")
+	}
 	var res uint32
 	err := raw.bus.Call("Notify", bus.Args(app_name, reuse_id, icon,
 		summary, body, actions, hints, timeout), &res)

=== modified file 'bus/notifications/raw_test.go'
--- bus/notifications/raw_test.go	2014-02-05 18:17:26 +0000
+++ bus/notifications/raw_test.go	2014-06-05 12:03:59 +0000
@@ -57,6 +57,12 @@
 	c.Check(err, NotNil)
 }
 
+func (s *RawSuite) TestNotifyFailsIfNoBus(c *C) {
+	raw := Raw(nil, s.log)
+	_, err := raw.Notify("", 0, "", "", "", nil, nil, 0)
+	c.Check(err, ErrorMatches, `.*unconfigured .*`)
+}
+
 func (s *RawSuite) TestNotifiesFailsWeirdly(c *C) {
 	endp := testibus.NewMultiValuedTestingEndpoint(nil, condition.Work(true), []interface{}{1, 2})
 	raw := Raw(endp, s.log)

=== modified file 'bus/testing/testing_endpoint.go'
--- bus/testing/testing_endpoint.go	2014-04-04 11:08:28 +0000
+++ bus/testing/testing_endpoint.go	2014-06-05 12:03:59 +0000
@@ -166,8 +166,46 @@
 		endp.dialCond, endp.callCond, endp.retvals)
 }
 
-// see Endpoint's Close. This one does nothing.
-func (tc *testingEndpoint) Close() {}
+// see Endpoint's Close. This one does nothing beyond registering
+// being called.
+func (tc *testingEndpoint) Close() {
+	tc.callArgsLck.Lock()
+	defer tc.callArgsLck.Unlock()
+
+	args := callArgs{Member: "::Close"}
+	tc.callArgs = append(tc.callArgs, args)
+}
+
+func (tc *testingEndpoint) GrabName(allowReplacement bool) <-chan error {
+	tc.callArgsLck.Lock()
+	defer tc.callArgsLck.Unlock()
+
+	args := callArgs{Member: "::GrabName"}
+	args.Args = append(args.Args, allowReplacement)
+	tc.callArgs = append(tc.callArgs, args)
+
+	return nil
+}
+
+func (tc *testingEndpoint) WatchMethod(dispatch bus.DispatchMap, extra ...interface{}) {
+	tc.callArgsLck.Lock()
+	defer tc.callArgsLck.Unlock()
+
+	args := callArgs{Member: "::WatchMethod"}
+	args.Args = append(args.Args, dispatch, extra)
+	tc.callArgs = append(tc.callArgs, args)
+}
+
+func (tc *testingEndpoint) Signal(member string, args []interface{}) error {
+	tc.callArgsLck.Lock()
+	defer tc.callArgsLck.Unlock()
+
+	callargs := callArgs{Member: "::Signal"}
+	callargs.Args = append(callargs.Args, member, args)
+	tc.callArgs = append(tc.callArgs, callargs)
+
+	return nil
+}
 
 // ensure testingEndpoint implements bus.Endpoint
-var _ bus.Endpoint = &testingEndpoint{}
+var _ bus.Endpoint = (*testingEndpoint)(nil)

=== modified file 'bus/testing/testing_endpoint_test.go'
--- bus/testing/testing_endpoint_test.go	2014-04-02 08:23:15 +0000
+++ bus/testing/testing_endpoint_test.go	2014-06-05 12:03:59 +0000
@@ -76,6 +76,24 @@
 		[]callArgs{{"what", []interface{}{"is", "this", "thing"}}})
 }
 
+// Test that Call() fails but does not explode when asked to return values
+// that can't be packed into a dbus message.
+func (s *TestingEndpointSuite) TestCallFailsOnBadRetval(c *C) {
+	endp := NewTestingEndpoint(nil, condition.Work(true), Equals)
+	var r uint32
+	e := endp.Call("what", bus.Args(), &r)
+	c.Check(e, NotNil)
+}
+
+// Test that Call() fails but does not explode when given an improper result
+// destination (one into which the dbus response can't be stuffed).
+func (s *TestingEndpointSuite) TestCallFailsOnBadArg(c *C) {
+	endp := NewTestingEndpoint(nil, condition.Work(true), 1)
+	r := func() {}
+	e := endp.Call("what", bus.Args(), &r)
+	c.Check(e, NotNil)
+}
+
 // Test that WatchSignal() with a positive condition sends the provided return
 // values over the channel.
 func (s *TestingEndpointSuite) TestWatch(c *C) {
@@ -102,7 +120,11 @@
 func (s *TestingEndpointSuite) TestCloser(c *C) {
 	endp := NewTestingEndpoint(nil, condition.Work(true))
 	endp.Close()
-	// ... yay?
+	c.Check(GetCallArgs(endp), DeepEquals, []callArgs{
+		{
+			Member: "::Close",
+			Args:   nil,
+		}})
 }
 
 // Test that WatchSignal() with a negative condition returns an error.
@@ -173,3 +195,46 @@
 	endp := NewTestingEndpoint(condition.Fail2Work(2), nil, "hello there")
 	c.Check(endp.String(), Matches, ".*Still Broken.*hello there.*")
 }
+
+// Test that GrabName updates callArgs
+func (s *TestingEndpointSuite) TestGrabNameUpdatesCallArgs(c *C) {
+	endp := NewTestingEndpoint(nil, condition.Work(true))
+	endp.GrabName(false)
+	endp.GrabName(true)
+	c.Check(GetCallArgs(endp), DeepEquals, []callArgs{
+		{
+			Member: "::GrabName",
+			Args:   []interface{}{false},
+		}, {
+			Member: "::GrabName",
+			Args:   []interface{}{true},
+		}})
+}
+
+// Test that Signal updates callArgs
+func (s *TestingEndpointSuite) TestSignalUpdatesCallArgs(c *C) {
+	endp := NewTestingEndpoint(nil, condition.Work(true))
+	endp.Signal("hello", []interface{}{"world"})
+	endp.Signal("hello", []interface{}{"there"})
+	c.Check(GetCallArgs(endp), DeepEquals, []callArgs{
+		{
+			Member: "::Signal",
+			Args:   []interface{}{"hello", []interface{}{"world"}},
+		}, {
+			Member: "::Signal",
+			Args:   []interface{}{"hello", []interface{}{"there"}},
+		}})
+}
+
+// Test that WatchMethod updates callArgs
+func (s *TestingEndpointSuite) TestWatchMethodUpdatesCallArgs(c *C) {
+	endp := NewTestingEndpoint(nil, condition.Work(true))
+	foo := func([]interface{}, []interface{}) ([]interface{}, error) { return nil, nil }
+	foomp := bus.DispatchMap{"foo": foo}
+	endp.WatchMethod(foomp)
+	c.Check(GetCallArgs(endp), DeepEquals, []callArgs{
+		{
+			Member: "::WatchMethod",
+			Args:   []interface{}{foomp, []interface{}(nil)},
+		}})
+}

=== modified file 'client/client.go'
--- client/client.go	2014-04-18 09:35:59 +0000
+++ client/client.go	2014-06-05 12:03:59 +0000
@@ -19,6 +19,10 @@
 package client
 
 import (
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
 	"encoding/pem"
 	"errors"
 	"fmt"
@@ -34,10 +38,12 @@
 	"launchpad.net/ubuntu-push/bus/notifications"
 	"launchpad.net/ubuntu-push/bus/systemimage"
 	"launchpad.net/ubuntu-push/bus/urldispatcher"
+	"launchpad.net/ubuntu-push/client/service"
 	"launchpad.net/ubuntu-push/client/session"
-	"launchpad.net/ubuntu-push/client/session/levelmap"
+	"launchpad.net/ubuntu-push/client/session/seenstate"
 	"launchpad.net/ubuntu-push/config"
 	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/util"
 	"launchpad.net/ubuntu-push/whoopsie/identifier"
 )
@@ -56,6 +62,8 @@
 	ExpectAllRepairedTime  config.ConfigTimeDuration `json:"expect_all_repaired"` // worth retrying all servers after
 	// The PEM-encoded server certificate
 	CertPEMFile string `json:"cert_pem_file"`
+	// How to invoke the auth helper
+	AuthHelper []string `json:"auth_helper"`
 	// The logging level (one of "debug", "info", "error")
 	LogLevel logger.ConfigLogLevel `json:"log_level"`
 }
@@ -79,9 +87,15 @@
 	actionsCh          <-chan notifications.RawActionReply
 	session            *session.ClientSession
 	sessionConnectedCh chan uint32
+	serviceEndpoint    bus.Endpoint
+	service            *service.Service
 }
 
-var ACTION_ID_SNOWFLAKE = "::ubuntu-push-client::"
+var (
+	system_update_url   = "settings:///system/system-update"
+	ACTION_ID_SNOWFLAKE = "::ubuntu-push-client::"
+	ACTION_ID_BROADCAST = ACTION_ID_SNOWFLAKE + system_update_url
+)
 
 // Creates a new Ubuntu Push Notifications client-side daemon that will use
 // the given configuration file.
@@ -144,8 +158,9 @@
 		ExchangeTimeout:        client.config.ExchangeTimeout.TimeDuration(),
 		HostsCachingExpiryTime: client.config.HostsCachingExpiryTime.TimeDuration(),
 		ExpectAllRepairedTime:  client.config.ExpectAllRepairedTime.TimeDuration(),
-		PEM:  client.pem,
-		Info: info,
+		PEM:        client.pem,
+		Info:       info,
+		AuthHelper: client.config.AuthHelper,
 	}
 }
 
@@ -155,7 +170,13 @@
 	if err != nil {
 		return err
 	}
-	client.deviceId = client.idder.String()
+	baseId := client.idder.String()
+	b, err := hex.DecodeString(baseId)
+	if err != nil {
+		return fmt.Errorf("whoopsie id should be hex: %v", err)
+	}
+	h := sha256.Sum224(b)
+	client.deviceId = base64.StdEncoding.EncodeToString(h[:])
 	return nil
 }
 
@@ -192,7 +213,7 @@
 	}
 	sess, err := session.NewSession(client.config.Addr,
 		client.deriveSessionConfig(info), client.deviceId,
-		client.levelMapFactory, client.log)
+		client.seenStateFactory, client.log)
 	if err != nil {
 		return err
 	}
@@ -200,12 +221,12 @@
 	return nil
 }
 
-// levelmapFactory returns a levelMap for the session
-func (client *PushClient) levelMapFactory() (levelmap.LevelMap, error) {
+// seenStateFactory returns a SeenState for the session
+func (client *PushClient) seenStateFactory() (seenstate.SeenState, error) {
 	if client.leveldbPath == "" {
-		return levelmap.NewLevelMap()
+		return seenstate.NewSeenState()
 	} else {
-		return levelmap.NewSqliteLevelMap(client.leveldbPath)
+		return seenstate.NewSqliteSeenState(client.leveldbPath)
 	}
 }
 
@@ -232,7 +253,7 @@
 	}
 }
 
-// filterNotification finds out if the notification is about an actual
+// filterBroadcastNotification finds out if the notification is about an actual
 // upgrade for the device. It expects msg.Decoded entries to look
 // like:
 //
@@ -240,7 +261,7 @@
 // "IMAGE-CHANNEL/DEVICE-MODEL": [BUILD-NUMBER, CHANNEL-ALIAS]
 // ...
 // }
-func (client *PushClient) filterNotification(msg *session.Notification) bool {
+func (client *PushClient) filterBroadcastNotification(msg *session.BroadcastNotification) bool {
 	n := len(msg.Decoded)
 	if n == 0 {
 		return false
@@ -275,26 +296,30 @@
 	return false
 }
 
-// handleNotification deals with receiving a notification
-func (client *PushClient) handleNotification(msg *session.Notification) error {
-	if !client.filterNotification(msg) {
-		return nil
-	}
-	action_id := ACTION_ID_SNOWFLAKE
-	a := []string{action_id, "Go get it!"} // action value not visible on the phone
+func (client *PushClient) sendNotification(action_id, icon, summary, body string) (uint32, error) {
+	a := []string{action_id, "Switch to app"} // action value not visible on the phone
 	h := map[string]*dbus.Variant{"x-canonical-switch-to-application": &dbus.Variant{true}}
 	nots := notifications.Raw(client.notificationsEndp, client.log)
-	body := "Tap to open the system updater."
-	not_id, err := nots.Notify(
-		"ubuntu-push-client",               // app name
-		uint32(0),                          // id
-		"update_manager_icon",              // icon
-		"There's an updated system image.", // summary
-		body,           // body
-		a,              // actions
-		h,              // hints
-		int32(10*1000), // timeout (ms)
+	return nots.Notify(
+		"ubuntu-push-client", // app name
+		uint32(0),            // id
+		icon,                 // icon
+		summary,              // summary
+		body,                 // body
+		a,                    // actions
+		h,                    // hints
+		int32(10*1000),       // timeout (ms)
 	)
+}
+
+// handleBroadcastNotification deals with receiving a broadcast notification
+func (client *PushClient) handleBroadcastNotification(msg *session.BroadcastNotification) error {
+	if !client.filterBroadcastNotification(msg) {
+		return nil
+	}
+	not_id, err := client.sendNotification(ACTION_ID_BROADCAST,
+		"update_manager_icon", "There's an updated system image.",
+		"Tap to open the system updater.")
 	if err != nil {
 		client.log.Errorf("showing notification: %s", err)
 		return err
@@ -303,26 +328,42 @@
 	return nil
 }
 
+// handleUnicastNotification deals with receiving a unicast notification
+func (client *PushClient) handleUnicastNotification(msg *protocol.Notification) error {
+	client.log.Debugf("sending notification %#v for %#v.", msg.MsgId, msg.AppId)
+	return client.service.Inject(msg.AppId, string(msg.Payload))
+}
+
 // handleClick deals with the user clicking a notification
 func (client *PushClient) handleClick(action_id string) error {
-	if action_id != ACTION_ID_SNOWFLAKE {
+	// “The string is a stark data structure and everywhere it is passed
+	// there is much duplication of process. It is a perfect vehicle for
+	// hiding information.”
+	//
+	// From ACM's SIGPLAN publication, (September, 1982), Article
+	// "Epigrams in Programming", by Alan J. Perlis of Yale University.
+	url := strings.TrimPrefix(action_id, ACTION_ID_SNOWFLAKE)
+	if len(url) == len(action_id) || len(url) == 0 {
+		// it didn't start with the prefix
 		return nil
 	}
 	// it doesn't get much simpler...
 	urld := urldispatcher.New(client.urlDispatcherEndp, client.log)
-	return urld.DispatchURL("settings:///system/system-update")
+	return urld.DispatchURL(url)
 }
 
 // doLoop connects events with their handlers
-func (client *PushClient) doLoop(connhandler func(bool), clickhandler func(string) error, notifhandler func(*session.Notification) error, errhandler func(error)) {
+func (client *PushClient) doLoop(connhandler func(bool), clickhandler func(string) error, bcasthandler func(*session.BroadcastNotification) error, ucasthandler func(*protocol.Notification) error, errhandler func(error)) {
 	for {
 		select {
 		case state := <-client.connCh:
 			connhandler(state)
 		case action := <-client.actionsCh:
 			clickhandler(action.ActionId)
-		case msg := <-client.session.MsgCh:
-			notifhandler(msg)
+		case bcast := <-client.session.BroadcastCh:
+			bcasthandler(bcast)
+		case ucast := <-client.session.NotificationsCh:
+			ucasthandler(ucast)
 		case err := <-client.session.ErrCh:
 			errhandler(err)
 		case count := <-client.sessionConnectedCh:
@@ -344,14 +385,57 @@
 
 // Loop calls doLoop with the "real" handlers
 func (client *PushClient) Loop() {
-	client.doLoop(client.handleConnState, client.handleClick,
-		client.handleNotification, client.handleErr)
+	client.doLoop(client.handleConnState,
+		client.handleClick,
+		client.handleBroadcastNotification,
+		client.handleUnicastNotification,
+		client.handleErr)
+}
+
+// these are the currently supported fields of a unicast message
+type UnicastMessage struct {
+	Icon    string          `json:"icon"`
+	Body    string          `json:"body"`
+	Summary string          `json:"summary"`
+	URL     string          `json:"url"`
+	Blob    json.RawMessage `json:"blob"`
+}
+
+func (client *PushClient) messageHandler(message []byte) error {
+	var umsg = new(UnicastMessage)
+	err := json.Unmarshal(message, &umsg)
+	if err != nil {
+		client.log.Errorf("unable to unmarshal message: %v", err)
+		return err
+	}
+
+	not_id, err := client.sendNotification(
+		ACTION_ID_SNOWFLAKE+umsg.URL,
+		umsg.Icon, umsg.Summary, umsg.Body)
+
+	if err != nil {
+		client.log.Errorf("showing notification: %s", err)
+		return err
+	}
+	client.log.Debugf("got notification id %d", not_id)
+	return nil
+}
+
+func (client *PushClient) startService() error {
+	if client.serviceEndpoint == nil {
+		client.serviceEndpoint = bus.SessionBus.Endpoint(service.BusAddress, client.log)
+	}
+
+	client.service = service.NewService(client.serviceEndpoint, client.log)
+	client.service.SetMessageHandler(client.messageHandler)
+	return client.service.Start()
 }
 
 // Start calls doStart with the "real" starters
 func (client *PushClient) Start() error {
 	return client.doStart(
 		client.configure,
+		client.startService,
 		client.getDeviceId,
 		client.takeTheBus,
 		client.initSession,

=== modified file 'client/client_test.go'
--- client/client_test.go	2014-04-18 16:31:04 +0000
+++ client/client_test.go	2014-06-05 12:03:59 +0000
@@ -38,8 +38,9 @@
 	"launchpad.net/ubuntu-push/bus/systemimage"
 	testibus "launchpad.net/ubuntu-push/bus/testing"
 	"launchpad.net/ubuntu-push/client/session"
-	"launchpad.net/ubuntu-push/client/session/levelmap"
+	"launchpad.net/ubuntu-push/client/session/seenstate"
 	"launchpad.net/ubuntu-push/config"
+	"launchpad.net/ubuntu-push/protocol"
 	helpers "launchpad.net/ubuntu-push/testing"
 	"launchpad.net/ubuntu-push/testing/condition"
 	"launchpad.net/ubuntu-push/util"
@@ -105,6 +106,7 @@
 		"addr":            ":0",
 		"cert_pem_file":   pem_file,
 		"recheck_timeout": "3h",
+		"auth_helper":     []string{},
 		"log_level":       "debug",
 	}
 	for k, v := range overrides {
@@ -254,6 +256,9 @@
 ******************************************************************/
 
 func (cs *clientSuite) TestDeriveSessionConfig(c *C) {
+	cs.writeTestConfig(map[string]interface{}{
+		"auth_helper": []string{"auth", "helper"},
+	})
 	info := map[string]interface{}{
 		"foo": 1,
 	}
@@ -265,8 +270,9 @@
 		ExchangeTimeout:        10 * time.Millisecond,
 		HostsCachingExpiryTime: 1 * time.Hour,
 		ExpectAllRepairedTime:  30 * time.Minute,
-		PEM:  cli.pem,
-		Info: info,
+		PEM:        cli.pem,
+		Info:       info,
+		AuthHelper: []string{"auth", "helper"},
 	}
 	// sanity check that we are looking at all fields
 	vExpected := reflect.ValueOf(expected)
@@ -282,6 +288,35 @@
 }
 
 /*****************************************************************
+    startService tests
+******************************************************************/
+
+func (cs *clientSuite) TestStartServiceWorks(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	cli.log = cs.log
+	cli.serviceEndpoint = testibus.NewTestingEndpoint(condition.Work(true), nil)
+	c.Check(cli.service, IsNil)
+	c.Check(cli.startService(), IsNil)
+	c.Assert(cli.service, NotNil)
+	c.Check(cli.service.IsRunning(), Equals, true)
+	c.Check(cli.service.GetMessageHandler(), NotNil)
+	cli.service.Stop()
+}
+
+func (cs *clientSuite) TestStartServiceErrorsOnNilLog(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	c.Check(cli.log, IsNil)
+	c.Check(cli.startService(), NotNil)
+}
+
+func (cs *clientSuite) TestStartServiceErrorsOnBusDialFail(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	cli.log = cs.log
+	cli.serviceEndpoint = testibus.NewTestingEndpoint(condition.Work(false), nil)
+	c.Check(cli.startService(), NotNil)
+}
+
+/*****************************************************************
     getDeviceId tests
 ******************************************************************/
 
@@ -291,7 +326,7 @@
 	cli.idder = identifier.New()
 	c.Check(cli.deviceId, Equals, "")
 	c.Check(cli.getDeviceId(), IsNil)
-	c.Check(cli.deviceId, HasLen, 128)
+	c.Check(cli.deviceId, HasLen, 40)
 }
 
 func (cs *clientSuite) TestGetDeviceIdCanFail(c *C) {
@@ -302,6 +337,16 @@
 	c.Check(cli.getDeviceId(), NotNil)
 }
 
+func (cs *clientSuite) TestGetDeviceIdWhoopsieDoesTheUnexpected(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	cli.log = cs.log
+	settable := idtesting.Settable()
+	cli.idder = settable
+	settable.Set("not-hex")
+	c.Check(cli.deviceId, Equals, "")
+	c.Check(cli.getDeviceId(), ErrorMatches, "whoopsie id should be hex: .*")
+}
+
 /*****************************************************************
     takeTheBus tests
 ******************************************************************/
@@ -389,21 +434,21 @@
 }
 
 /*****************************************************************
-    levelmapFactory tests
+    seenStateFactory tests
 ******************************************************************/
 
-func (cs *clientSuite) TestLevelMapFactoryNoDbPath(c *C) {
+func (cs *clientSuite) TestSeenStateFactoryNoDbPath(c *C) {
 	cli := NewPushClient(cs.configPath, "")
-	ln, err := cli.levelMapFactory()
+	ln, err := cli.seenStateFactory()
 	c.Assert(err, IsNil)
-	c.Check(fmt.Sprintf("%T", ln), Equals, "*levelmap.mapLevelMap")
+	c.Check(fmt.Sprintf("%T", ln), Equals, "*seenstate.memSeenState")
 }
 
-func (cs *clientSuite) TestLevelMapFactoryWithDbPath(c *C) {
+func (cs *clientSuite) TestSeenStateFactoryWithDbPath(c *C) {
 	cli := NewPushClient(cs.configPath, ":memory:")
-	ln, err := cli.levelMapFactory()
+	ln, err := cli.seenStateFactory()
 	c.Assert(err, IsNil)
-	c.Check(fmt.Sprintf("%T", ln), Equals, "*levelmap.sqliteLevelMap")
+	c.Check(fmt.Sprintf("%T", ln), Equals, "*seenstate.sqliteSeenState")
 }
 
 /*****************************************************************
@@ -439,7 +484,7 @@
 func (cs *clientSuite) TestHandleConnStateC2D(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.log = cs.log
-	cli.session, _ = session.NewSession(cli.config.Addr, cli.deriveSessionConfig(nil), cli.deviceId, levelmap.NewLevelMap, cs.log)
+	cli.session, _ = session.NewSession(cli.config.Addr, cli.deriveSessionConfig(nil), cli.deviceId, seenstate.NewSeenState, cs.log)
 	cli.session.Dial()
 	cli.hasConnectivity = true
 
@@ -452,7 +497,7 @@
 func (cs *clientSuite) TestHandleConnStateC2DPending(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.log = cs.log
-	cli.session, _ = session.NewSession(cli.config.Addr, cli.deriveSessionConfig(nil), cli.deviceId, levelmap.NewLevelMap, cs.log)
+	cli.session, _ = session.NewSession(cli.config.Addr, cli.deriveSessionConfig(nil), cli.deviceId, seenstate.NewSeenState, cs.log)
 	cli.hasConnectivity = true
 
 	cli.handleConnState(false)
@@ -460,7 +505,7 @@
 }
 
 /*****************************************************************
-   filterNotification tests
+   filterBroadcastNotification tests
 ******************************************************************/
 
 var siInfoRes = &systemimage.InfoResult{
@@ -470,23 +515,23 @@
 	LastUpdate:  "Unknown",
 }
 
-func (cs *clientSuite) TestFilterNotification(c *C) {
+func (cs *clientSuite) TestFilterBroadcastNotification(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	// empty
-	msg := &session.Notification{}
-	c.Check(cli.filterNotification(msg), Equals, false)
+	msg := &session.BroadcastNotification{}
+	c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	// same build number
-	msg = &session.Notification{
+	msg = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
 			},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, false)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	// higher build number and pick last
-	msg = &session.Notification{
+	msg = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
@@ -496,9 +541,9 @@
 			},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, true)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, true)
 	// going backward by a margin, assume switch of alias
-	msg = &session.Notification{
+	msg = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
@@ -508,47 +553,47 @@
 			},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, true)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, true)
 }
 
-func (cs *clientSuite) TestFilterNotificationRobust(c *C) {
+func (cs *clientSuite) TestFilterBroadcastNotificationRobust(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
-	msg := &session.Notification{
+	msg := &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{},
 		},
 	}
-	c.Check(cli.filterNotification(msg), Equals, false)
+	c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	for _, broken := range []interface{}{
 		5,
 		[]interface{}{},
 		[]interface{}{55},
 	} {
-		msg := &session.Notification{
+		msg := &session.BroadcastNotification{
 			Decoded: []map[string]interface{}{
 				map[string]interface{}{
 					"daily/mako": broken,
 				},
 			},
 		}
-		c.Check(cli.filterNotification(msg), Equals, false)
+		c.Check(cli.filterBroadcastNotification(msg), Equals, false)
 	}
 }
 
 /*****************************************************************
-    handleNotification tests
+    handleBroadcastNotification tests
 ******************************************************************/
 
 var (
-	positiveNotification = &session.Notification{
+	positiveBroadcastNotification = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(103), "tubular"},
 			},
 		},
 	}
-	negativeNotification = &session.Notification{
+	negativeBroadcastNotification = &session.BroadcastNotification{
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
 				"daily/mako": []interface{}{float64(102), "tubular"},
@@ -557,13 +602,13 @@
 	}
 )
 
-func (cs *clientSuite) TestHandleNotification(c *C) {
+func (cs *clientSuite) TestHandleBroadcastNotification(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(true), uint32(1))
 	cli.notificationsEndp = endp
 	cli.log = cs.log
-	c.Check(cli.handleNotification(positiveNotification), IsNil)
+	c.Check(cli.handleBroadcastNotification(positiveBroadcastNotification), IsNil)
 	// check we sent the notification
 	args := testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 1)
@@ -571,26 +616,48 @@
 	c.Check(cs.log.Captured(), Matches, `.* got notification id \d+\s*`)
 }
 
-func (cs *clientSuite) TestHandleNotificationNothingToDo(c *C) {
+func (cs *clientSuite) TestHandleBroadcastNotificationNothingToDo(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(true), uint32(1))
 	cli.notificationsEndp = endp
 	cli.log = cs.log
-	c.Check(cli.handleNotification(negativeNotification), IsNil)
+	c.Check(cli.handleBroadcastNotification(negativeBroadcastNotification), IsNil)
 	// check we sent the notification
 	args := testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 0)
 	c.Check(cs.log.Captured(), Matches, "")
 }
 
-func (cs *clientSuite) TestHandleNotificationFail(c *C) {
+func (cs *clientSuite) TestHandleBroadcastNotificationFail(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.systemImageInfo = siInfoRes
 	cli.log = cs.log
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(false))
 	cli.notificationsEndp = endp
-	c.Check(cli.handleNotification(positiveNotification), NotNil)
+	c.Check(cli.handleBroadcastNotification(positiveBroadcastNotification), NotNil)
+}
+
+/*****************************************************************
+    handleUnicastNotification tests
+******************************************************************/
+
+var notif = &protocol.Notification{AppId: "hello", Payload: []byte(`{"url": "xyzzy"}`), MsgId: "42"}
+
+func (cs *clientSuite) TestHandleUcastNotification(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	svcEndp := testibus.NewTestingEndpoint(condition.Work(true), condition.Work(true), uint32(1))
+	cli.log = cs.log
+	cli.serviceEndpoint = svcEndp
+	notsEndp := testibus.NewTestingEndpoint(nil, condition.Work(true), uint32(1))
+	cli.notificationsEndp = notsEndp
+	c.Assert(cli.startService(), IsNil)
+	c.Check(cli.handleUnicastNotification(notif), IsNil)
+	// check we sent the notification
+	args := testibus.GetCallArgs(svcEndp)
+	c.Assert(len(args), Not(Equals), 0)
+	c.Check(args[len(args)-1].Member, Equals, "::Signal")
+	c.Check(cs.log.Captured(), Matches, `(?m).*sending notification "42" for "hello".*`)
 }
 
 /*****************************************************************
@@ -608,18 +675,31 @@
 	args := testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 0)
 	// check we worked with the right action id
-	c.Check(cli.handleClick(ACTION_ID_SNOWFLAKE), IsNil)
+	c.Check(cli.handleClick(ACTION_ID_BROADCAST), IsNil)
 	// check we sent the notification
 	args = testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 1)
 	c.Check(args[0].Member, Equals, "DispatchURL")
-	c.Check(args[0].Args, DeepEquals, []interface{}{"settings:///system/system-update"})
+	c.Check(args[0].Args, DeepEquals, []interface{}{system_update_url})
+	// check we worked with the right action id
+	c.Check(cli.handleClick(ACTION_ID_SNOWFLAKE+"foo"), IsNil)
+	// check we sent the notification
+	args = testibus.GetCallArgs(endp)
+	c.Assert(args, HasLen, 2)
+	c.Check(args[1].Member, Equals, "DispatchURL")
+	c.Check(args[1].Args, DeepEquals, []interface{}{"foo"})
 }
 
 /*****************************************************************
     doLoop tests
 ******************************************************************/
 
+var nopConn = func(bool) {}
+var nopClick = func(string) error { return nil }
+var nopBcast = func(*session.BroadcastNotification) error { return nil }
+var nopUcast = func(*protocol.Notification) error { return nil }
+var nopError = func(error) {}
+
 func (cs *clientSuite) TestDoLoopConn(c *C) {
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.log = cs.log
@@ -629,7 +709,7 @@
 	c.Assert(cli.initSession(), IsNil)
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) { ch <- true }, func(_ string) error { return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
+	go cli.doLoop(func(bool) { ch <- true }, nopClick, nopBcast, nopUcast, nopError)
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -643,7 +723,20 @@
 	cli.actionsCh = aCh
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func(_ string) error { ch <- true; return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
+	go cli.doLoop(nopConn, func(_ string) error { ch <- true; return nil }, nopBcast, nopUcast, nopError)
+	c.Check(takeNextBool(ch), Equals, true)
+}
+
+func (cs *clientSuite) TestDoLoopBroadcast(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	cli.log = cs.log
+	cli.systemImageInfo = siInfoRes
+	c.Assert(cli.initSession(), IsNil)
+	cli.session.BroadcastCh = make(chan *session.BroadcastNotification, 1)
+	cli.session.BroadcastCh <- &session.BroadcastNotification{}
+
+	ch := make(chan bool, 1)
+	go cli.doLoop(nopConn, nopClick, func(_ *session.BroadcastNotification) error { ch <- true; return nil }, nopUcast, nopError)
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -652,11 +745,11 @@
 	cli.log = cs.log
 	cli.systemImageInfo = siInfoRes
 	c.Assert(cli.initSession(), IsNil)
-	cli.session.MsgCh = make(chan *session.Notification, 1)
-	cli.session.MsgCh <- &session.Notification{}
+	cli.session.NotificationsCh = make(chan *protocol.Notification, 1)
+	cli.session.NotificationsCh <- &protocol.Notification{}
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.Notification) error { ch <- true; return nil }, func(error) {})
+	go cli.doLoop(nopConn, nopClick, nopBcast, func(*protocol.Notification) error { ch <- true; return nil }, nopError)
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -669,7 +762,7 @@
 	cli.session.ErrCh <- nil
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.Notification) error { return nil }, func(error) { ch <- true })
+	go cli.doLoop(nopConn, nopClick, nopBcast, nopUcast, func(error) { ch <- true })
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -719,7 +812,7 @@
 	cli.systemImageInfo = siInfoRes
 	c.Assert(cli.initSession(), IsNil)
 
-	cli.session.MsgCh = make(chan *session.Notification)
+	cli.session.BroadcastCh = make(chan *session.BroadcastNotification)
 	cli.session.ErrCh = make(chan error)
 
 	// we use tick() to make sure things have been through the
@@ -736,7 +829,7 @@
 	c.Check(cs.log.Captured(), Matches, "(?ms).*Session connected after 42 attempts$")
 
 	//  * actionsCh to the click handler/url dispatcher
-	aCh <- notifications.RawActionReply{ActionId: ACTION_ID_SNOWFLAKE}
+	aCh <- notifications.RawActionReply{ActionId: ACTION_ID_BROADCAST}
 	tick()
 	uargs := testibus.GetCallArgs(cli.urlDispatcherEndp)
 	c.Assert(uargs, HasLen, 1)
@@ -752,8 +845,8 @@
 	tick()
 	c.Check(cli.hasConnectivity, Equals, false)
 
-	//  * session.MsgCh to the notifications handler
-	cli.session.MsgCh <- positiveNotification
+	//  * session.BroadcastCh to the notifications handler
+	cli.session.BroadcastCh <- positiveBroadcastNotification
 	tick()
 	nargs := testibus.GetCallArgs(cli.notificationsEndp)
 	c.Check(nargs, HasLen, 1)
@@ -785,6 +878,8 @@
 
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	// before start, everything sucks:
+	// no service,
+	c.Check(cli.service, IsNil)
 	// no config,
 	c.Check(string(cli.config.Addr), Equals, "")
 	// no device id,
@@ -803,12 +898,15 @@
 	// and now everthing is better! We have a config,
 	c.Check(string(cli.config.Addr), Equals, ":0")
 	// and a device id,
-	c.Check(cli.deviceId, HasLen, 128)
+	c.Check(cli.deviceId, HasLen, 40)
 	// and a session,
 	c.Check(cli.session, NotNil)
 	// and a bus,
 	c.Check(cli.notificationsEndp, NotNil)
+	// and a service,
+	c.Check(cli.service, NotNil)
 	// and everthying us just peachy!
+	cli.service.Stop() // cleanup
 }
 
 func (cs *clientSuite) TestStartCanFail(c *C) {
@@ -818,3 +916,38 @@
 	// and it works. Err. Doesn't.
 	c.Check(err, NotNil)
 }
+
+func (cs *clientSuite) TestMessageHandler(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	endp := testibus.NewTestingEndpoint(nil, condition.Work(true), uint32(1))
+	cli.notificationsEndp = endp
+	cli.log = cs.log
+	err := cli.messageHandler([]byte(`{"icon": "icon-value", "summary": "summary-value", "body": "body-value"}`))
+	c.Assert(err, IsNil)
+	args := testibus.GetCallArgs(endp)
+	c.Assert(args, HasLen, 1)
+	c.Check(args[0].Member, Equals, "Notify")
+	c.Check(args[0].Args[0], Equals, "ubuntu-push-client")
+	c.Check(args[0].Args[2], Equals, "icon-value")
+	c.Check(args[0].Args[3], Equals, "summary-value")
+	c.Check(args[0].Args[4], Equals, "body-value")
+}
+
+func (cs *clientSuite) TestMessageHandlerReportsUnmarshalErrors(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	cli.log = cs.log
+
+	err := cli.messageHandler([]byte(`{"broken`))
+	c.Check(err, NotNil)
+	c.Check(cs.log.Captured(), Matches, "(?msi).*unable to unmarshal message:.*")
+}
+
+func (cs *clientSuite) TestMessageHandlerReportsFailedNotifies(c *C) {
+	cli := NewPushClient(cs.configPath, cs.leveldbPath)
+	endp := testibus.NewTestingEndpoint(nil, condition.Work(false))
+	cli.notificationsEndp = endp
+	cli.log = cs.log
+	err := cli.messageHandler([]byte(`{}`))
+	c.Assert(err, NotNil)
+	c.Check(cs.log.Captured(), Matches, "(?msi).*showing notification: no way$")
+}

=== added directory 'client/service'
=== added file 'client/service/service.go'
--- client/service/service.go	1970-01-01 00:00:00 +0000
+++ client/service/service.go	2014-06-05 12:03:59 +0000
@@ -0,0 +1,209 @@
+/*
+ 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 service implements the dbus-level service with which client
+// applications are expected to interact.
+package service
+
+import (
+	"errors"
+	"os"
+	"sync"
+
+	"launchpad.net/ubuntu-push/bus"
+	"launchpad.net/ubuntu-push/logger"
+)
+
+// Service is the dbus api
+type Service struct {
+	lock       sync.RWMutex
+	state      ServiceState
+	mbox       map[string][]string
+	msgHandler func([]byte) error
+	Log        logger.Logger
+	Bus        bus.Endpoint
+}
+
+// the service can be in a numnber of states
+type ServiceState uint8
+
+const (
+	StateUnknown  ServiceState = iota
+	StateRunning               // Start() has been successfully called
+	StateFinished              // Stop() has been successfully called
+)
+
+var (
+	NotConfigured  = errors.New("not configured")
+	AlreadyStarted = errors.New("already started")
+	BusAddress     = bus.Address{
+		Interface: "com.ubuntu.PushNotifications",
+		Path:      "/com/ubuntu/PushNotifications",
+		Name:      "com.ubuntu.PushNotifications",
+	}
+)
+
+// NewService() builds a new service and returns it.
+func NewService(bus bus.Endpoint, log logger.Logger) *Service {
+	return &Service{Log: log, Bus: bus}
+}
+
+// SetMessageHandler() sets the message-handling callback
+func (svc *Service) SetMessageHandler(callback func([]byte) error) {
+	svc.lock.Lock()
+	defer svc.lock.Unlock()
+	svc.msgHandler = callback
+}
+
+// GetMessageHandler() returns the (possibly nil) messaging handler callback
+func (svc *Service) GetMessageHandler() func([]byte) error {
+	svc.lock.RLock()
+	defer svc.lock.RUnlock()
+	return svc.msgHandler
+}
+
+// IsRunning() returns whether the service's state is StateRunning
+func (svc *Service) IsRunning() bool {
+	svc.lock.RLock()
+	defer svc.lock.RUnlock()
+	return svc.state == StateRunning
+}
+
+// Start() dials the bus, grab the name, and listens for method calls.
+func (svc *Service) Start() error {
+	svc.lock.Lock()
+	defer svc.lock.Unlock()
+	if svc.state != StateUnknown {
+		return AlreadyStarted
+	}
+	if svc.Log == nil || svc.Bus == nil {
+		return NotConfigured
+	}
+	err := svc.Bus.Dial()
+	if err != nil {
+		return err
+	}
+	ch := svc.Bus.GrabName(true)
+	log := svc.Log
+	go func() {
+		for err := range ch {
+			if !svc.IsRunning() {
+				break
+			}
+			if err != nil {
+				log.Fatalf("name channel for %s got: %v",
+					BusAddress.Name, err)
+			}
+		}
+	}()
+	svc.Bus.WatchMethod(bus.DispatchMap{
+		"Register":      svc.register,
+		"Notifications": svc.notifications,
+		"Inject":        svc.inject,
+	}, svc)
+	svc.state = StateRunning
+	return nil
+}
+
+// Stop() closes the bus and sets the state to StateFinished
+func (svc *Service) Stop() {
+	svc.lock.Lock()
+	defer svc.lock.Unlock()
+	if svc.Bus != nil {
+		svc.Bus.Close()
+	}
+	svc.state = StateFinished
+}
+
+var (
+	BadArgCount = errors.New("Wrong number of arguments")
+	BadArgType  = errors.New("Bad argument type")
+)
+
+func (svc *Service) register(args []interface{}, _ []interface{}) ([]interface{}, error) {
+	if len(args) != 1 {
+		return nil, BadArgCount
+	}
+	appname, ok := args[0].(string)
+	if !ok {
+		return nil, BadArgType
+	}
+
+	rv := os.Getenv("PUSH_REG_" + appname)
+	if rv == "" {
+		rv = "this-is-an-opaque-block-of-random-bits-i-promise"
+	}
+
+	return []interface{}{rv}, nil
+}
+
+func (svc *Service) notifications(args []interface{}, _ []interface{}) ([]interface{}, error) {
+	if len(args) != 1 {
+		return nil, BadArgCount
+	}
+	appname, ok := args[0].(string)
+	if !ok {
+		return nil, BadArgType
+	}
+
+	svc.lock.Lock()
+	defer svc.lock.Unlock()
+
+	if svc.mbox == nil {
+		return []interface{}{[]string(nil)}, nil
+	}
+	msgs := svc.mbox[appname]
+	delete(svc.mbox, appname)
+
+	return []interface{}{msgs}, nil
+}
+
+func (svc *Service) inject(args []interface{}, _ []interface{}) ([]interface{}, error) {
+	if len(args) != 2 {
+		return nil, BadArgCount
+	}
+	appname, ok := args[0].(string)
+	if !ok {
+		return nil, BadArgType
+	}
+	notif, ok := args[1].(string)
+	if !ok {
+		return nil, BadArgType
+	}
+
+	return nil, svc.Inject(appname, notif)
+}
+
+// Inject() signals to an application over dbus that a notification
+// has arrived.
+func (svc *Service) Inject(appname string, notif string) error {
+	svc.lock.Lock()
+	defer svc.lock.Unlock()
+	if svc.mbox == nil {
+		svc.mbox = make(map[string][]string)
+	}
+	svc.mbox[appname] = append(svc.mbox[appname], notif)
+	if svc.msgHandler != nil {
+		err := svc.msgHandler([]byte(notif))
+		if err != nil {
+			svc.Log.Errorf("msgHandler returned %v", err)
+			return err
+		}
+		svc.Log.Debugf("call to msgHandler successful")
+	}
+
+	return svc.Bus.Signal("Notification", []interface{}{appname})
+}

=== added file 'client/service/service_test.go'
--- client/service/service_test.go	1970-01-01 00:00:00 +0000
+++ client/service/service_test.go	2014-06-05 12:03:59 +0000
@@ -0,0 +1,247 @@
+/*
+ Copyright 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 service
+
+import (
+	"errors"
+	"os"
+	"testing"
+
+	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/bus"
+	testibus "launchpad.net/ubuntu-push/bus/testing"
+	"launchpad.net/ubuntu-push/logger"
+	helpers "launchpad.net/ubuntu-push/testing"
+	"launchpad.net/ubuntu-push/testing/condition"
+)
+
+func TestService(t *testing.T) { TestingT(t) }
+
+type serviceSuite struct {
+	log logger.Logger
+	bus bus.Endpoint
+}
+
+var _ = Suite(&serviceSuite{})
+
+func (ss *serviceSuite) SetUpTest(c *C) {
+	ss.log = helpers.NewTestLogger(c, "debug")
+	ss.bus = testibus.NewTestingEndpoint(condition.Work(true), nil)
+}
+
+func (ss *serviceSuite) TestStart(c *C) {
+	svc := NewService(ss.bus, ss.log)
+	c.Check(svc.IsRunning(), Equals, false)
+	c.Check(svc.Start(), IsNil)
+	c.Check(svc.IsRunning(), Equals, true)
+	svc.Stop()
+}
+
+func (ss *serviceSuite) TestStartTwice(c *C) {
+	svc := NewService(ss.bus, ss.log)
+	c.Check(svc.Start(), IsNil)
+	c.Check(svc.Start(), Equals, AlreadyStarted)
+	svc.Stop()
+}
+
+func (ss *serviceSuite) TestStartNoLog(c *C) {
+	svc := NewService(ss.bus, nil)
+	c.Check(svc.Start(), Equals, NotConfigured)
+}
+
+func (ss *serviceSuite) TestStartNoBus(c *C) {
+	svc := NewService(nil, ss.log)
+	c.Check(svc.Start(), Equals, NotConfigured)
+}
+
+func (ss *serviceSuite) TestStartFailsOnBusDialFailure(c *C) {
+	bus := testibus.NewTestingEndpoint(condition.Work(false), nil)
+	svc := NewService(bus, ss.log)
+	c.Check(svc.Start(), ErrorMatches, `.*(?i)cond said no.*`)
+	svc.Stop()
+}
+
+func (ss *serviceSuite) TestStartGrabsName(c *C) {
+	svc := NewService(ss.bus, ss.log)
+	c.Assert(svc.Start(), IsNil)
+	callArgs := testibus.GetCallArgs(ss.bus)
+	defer svc.Stop()
+	c.Assert(callArgs, NotNil)
+	c.Check(callArgs[0].Member, Equals, "::GrabName")
+}
+
+func (ss *serviceSuite) TestStopClosesBus(c *C) {
+	svc := NewService(ss.bus, ss.log)
+	c.Assert(svc.Start(), IsNil)
+	svc.Stop()
+	callArgs := testibus.GetCallArgs(ss.bus)
+	c.Assert(callArgs, NotNil)
+	c.Check(callArgs[len(callArgs)-1].Member, Equals, "::Close")
+}
+
+// registration tests
+
+func (ss *serviceSuite) TestRegistrationFailsIfBadArgs(c *C) {
+	for i, s := range []struct {
+		args []interface{}
+		errt error
+	}{
+		{nil, BadArgCount},                 // no args
+		{[]interface{}{}, BadArgCount},     // still no args
+		{[]interface{}{42}, BadArgType},    // bad arg type
+		{[]interface{}{1, 2}, BadArgCount}, // too many args
+	} {
+		reg, err := new(Service).register(s.args, nil)
+		c.Check(reg, IsNil, Commentf("iteration #%d", i))
+		c.Check(err, Equals, s.errt, Commentf("iteration #%d", i))
+	}
+}
+
+func (ss *serviceSuite) TestRegistrationWorks(c *C) {
+	reg, err := new(Service).register([]interface{}{"this"}, nil)
+	c.Assert(reg, HasLen, 1)
+	regs, ok := reg[0].(string)
+	c.Check(ok, Equals, true)
+	c.Check(regs, Not(Equals), "")
+	c.Check(err, IsNil)
+}
+
+func (ss *serviceSuite) TestRegistrationOverrideWorks(c *C) {
+	os.Setenv("PUSH_REG_stuff", "42")
+	defer os.Setenv("PUSH_REG_stuff", "")
+
+	reg, err := new(Service).register([]interface{}{"stuff"}, nil)
+	c.Assert(reg, HasLen, 1)
+	regs, ok := reg[0].(string)
+	c.Check(ok, Equals, true)
+	c.Check(regs, Equals, "42")
+	c.Check(err, IsNil)
+}
+
+//
+// Injection tests
+
+func (ss *serviceSuite) TestInjectWorks(c *C) {
+	svc := NewService(ss.bus, ss.log)
+	rvs, err := svc.inject([]interface{}{"hello", "world"}, nil)
+	c.Assert(err, IsNil)
+	c.Check(rvs, IsNil)
+	rvs, err = svc.inject([]interface{}{"hello", "there"}, nil)
+	c.Assert(err, IsNil)
+	c.Check(rvs, IsNil)
+	c.Assert(svc.mbox, HasLen, 1)
+	c.Assert(svc.mbox["hello"], HasLen, 2)
+	c.Check(svc.mbox["hello"][0], Equals, "world")
+	c.Check(svc.mbox["hello"][1], Equals, "there")
+
+	// and check it fired the right signal (twice)
+	callArgs := testibus.GetCallArgs(ss.bus)
+	c.Assert(callArgs, HasLen, 2)
+	c.Check(callArgs[0].Member, Equals, "::Signal")
+	c.Check(callArgs[0].Args, DeepEquals, []interface{}{"Notification", []interface{}{"hello"}})
+	c.Check(callArgs[1], DeepEquals, callArgs[0])
+}
+
+func (ss *serviceSuite) TestInjectFailsIfInjectFails(c *C) {
+	bus := testibus.NewTestingEndpoint(condition.Work(true),
+		condition.Work(false))
+	svc := NewService(bus, ss.log)
+	svc.SetMessageHandler(func([]byte) error { return errors.New("fail") })
+	_, err := svc.inject([]interface{}{"hello", "xyzzy"}, nil)
+	c.Check(err, NotNil)
+}
+
+func (ss *serviceSuite) TestInjectFailsIfBadArgs(c *C) {
+	for i, s := range []struct {
+		args []interface{}
+		errt error
+	}{
+		{nil, BadArgCount},
+		{[]interface{}{}, BadArgCount},
+		{[]interface{}{1}, BadArgCount},
+		{[]interface{}{1, 2}, BadArgType},
+		{[]interface{}{"1", 2}, BadArgType},
+		{[]interface{}{1, "2"}, BadArgType},
+		{[]interface{}{1, 2, 3}, BadArgCount},
+	} {
+		reg, err := new(Service).inject(s.args, nil)
+		c.Check(reg, IsNil, Commentf("iteration #%d", i))
+		c.Check(err, Equals, s.errt, Commentf("iteration #%d", i))
+	}
+}
+
+//
+// Notifications tests
+func (ss *serviceSuite) TestNotificationsWorks(c *C) {
+	svc := NewService(ss.bus, ss.log)
+	nots, err := svc.notifications([]interface{}{"hello"}, nil)
+	c.Assert(err, IsNil)
+	c.Assert(nots, NotNil)
+	c.Assert(nots, HasLen, 1)
+	c.Check(nots[0], HasLen, 0)
+	if svc.mbox == nil {
+		svc.mbox = make(map[string][]string)
+	}
+	svc.mbox["hello"] = append(svc.mbox["hello"], "this", "thing")
+	nots, err = svc.notifications([]interface{}{"hello"}, nil)
+	c.Assert(err, IsNil)
+	c.Assert(nots, NotNil)
+	c.Assert(nots, HasLen, 1)
+	c.Check(nots[0], DeepEquals, []string{"this", "thing"})
+}
+
+func (ss *serviceSuite) TestNotificationsFailsIfBadArgs(c *C) {
+	for i, s := range []struct {
+		args []interface{}
+		errt error
+	}{
+		{nil, BadArgCount},                 // no args
+		{[]interface{}{}, BadArgCount},     // still no args
+		{[]interface{}{42}, BadArgType},    // bad arg type
+		{[]interface{}{1, 2}, BadArgCount}, // too many args
+	} {
+		reg, err := new(Service).notifications(s.args, nil)
+		c.Check(reg, IsNil, Commentf("iteration #%d", i))
+		c.Check(err, Equals, s.errt, Commentf("iteration #%d", i))
+	}
+}
+
+func (ss *serviceSuite) TestMessageHandler(c *C) {
+	svc := new(Service)
+	c.Assert(svc.msgHandler, IsNil)
+	var ext = []byte{}
+	e := errors.New("Hello")
+	f := func(s []byte) error { ext = s; return e }
+	c.Check(svc.GetMessageHandler(), IsNil)
+	svc.SetMessageHandler(f)
+	c.Check(svc.GetMessageHandler(), NotNil)
+	c.Check(svc.msgHandler([]byte("37")), Equals, e)
+	c.Check(ext, DeepEquals, []byte("37"))
+}
+
+func (ss *serviceSuite) TestInjectCallsMessageHandler(c *C) {
+	var ext = []byte{}
+	svc := NewService(ss.bus, ss.log)
+	f := func(s []byte) error { ext = s; return nil }
+	svc.SetMessageHandler(f)
+	c.Check(svc.Inject("stuff", "{}"), IsNil)
+	c.Check(ext, DeepEquals, []byte("{}"))
+	err := errors.New("ouch")
+	svc.SetMessageHandler(func([]byte) error { return err })
+	c.Check(svc.Inject("stuff", "{}"), Equals, err)
+}

=== renamed directory 'client/session/levelmap' => 'client/session/seenstate'
=== renamed file 'client/session/levelmap/levelmap.go' => 'client/session/seenstate/seenstate.go'
--- client/session/levelmap/levelmap.go	2014-02-21 16:17:28 +0000
+++ client/session/seenstate/seenstate.go	2014-06-05 12:03:59 +0000
@@ -14,31 +14,57 @@
  with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-// Package levelmap holds implementations of the LevelMap that the client
+// Package seenstate holds implementations of the SeenState that the client
 // session uses to keep track of what messages it has seen.
-package levelmap
-
-type LevelMap interface {
+package seenstate
+
+import (
+	"launchpad.net/ubuntu-push/protocol"
+)
+
+type SeenState interface {
 	// Set() (re)sets the given level to the given value.
-	Set(level string, top int64) error
+	SetLevel(level string, top int64) error
 	// GetAll() returns a "simple" map of the current levels.
-	GetAll() (map[string]int64, error)
-}
-
-type mapLevelMap map[string]int64
-
-func (m *mapLevelMap) Set(level string, top int64) error {
-	(*m)[level] = top
+	GetAllLevels() (map[string]int64, error)
+	// FilterBySeen filters notifications already seen, keep track
+	// of them as well
+	FilterBySeen([]protocol.Notification) ([]protocol.Notification, error)
+}
+
+type memSeenState struct {
+	levels   map[string]int64
+	seenMsgs map[string]bool
+}
+
+func (m *memSeenState) SetLevel(level string, top int64) error {
+	m.levels[level] = top
 	return nil
 }
-func (m *mapLevelMap) GetAll() (map[string]int64, error) {
-	return map[string]int64(*m), nil
-}
-
-var _ LevelMap = &mapLevelMap{}
-
-// NewLevelMap returns an implementation of LevelMap that is memory-based and
+func (m *memSeenState) GetAllLevels() (map[string]int64, error) {
+	return m.levels, nil
+}
+
+func (m *memSeenState) FilterBySeen(notifs []protocol.Notification) ([]protocol.Notification, error) {
+	acc := make([]protocol.Notification, 0, len(notifs))
+	for _, notif := range notifs {
+		seen := m.seenMsgs[notif.MsgId]
+		if seen {
+			continue
+		}
+		m.seenMsgs[notif.MsgId] = true
+		acc = append(acc, notif)
+	}
+	return acc, nil
+}
+
+var _ SeenState = (*memSeenState)(nil)
+
+// NewSeenState returns an implementation of SeenState that is memory-based and
 // does not save state.
-func NewLevelMap() (LevelMap, error) {
-	return &mapLevelMap{}, nil
+func NewSeenState() (SeenState, error) {
+	return &memSeenState{
+		levels:   make(map[string]int64),
+		seenMsgs: make(map[string]bool),
+	}, nil
 }

=== renamed file 'client/session/levelmap/levelmap_test.go' => 'client/session/seenstate/seenstate_test.go'
--- client/session/levelmap/levelmap_test.go	2014-02-08 13:50:58 +0000
+++ client/session/seenstate/seenstate_test.go	2014-06-05 12:03:59 +0000
@@ -14,42 +14,73 @@
  with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-package levelmap
+package seenstate
 
 import (
+	"testing"
+
 	. "launchpad.net/gocheck"
-	"testing"
+
+	"launchpad.net/ubuntu-push/protocol"
 )
 
-func TestLevelMap(t *testing.T) { TestingT(t) }
-
-type lmSuite struct {
-	constructor func() (LevelMap, error)
-}
-
-var _ = Suite(&lmSuite{})
-
-func (s *lmSuite) SetUpSuite(c *C) {
-	s.constructor = NewLevelMap
-}
-
-func (s *lmSuite) TestAllTheThings(c *C) {
+func TestSeenState(t *testing.T) { TestingT(t) }
+
+type ssSuite struct {
+	constructor func() (SeenState, error)
+}
+
+var _ = Suite(&ssSuite{})
+
+func (s *ssSuite) SetUpSuite(c *C) {
+	s.constructor = NewSeenState
+}
+
+func (s *ssSuite) TestAllTheLevelThings(c *C) {
 	var err error
-	var lm LevelMap
-	// checks NewLevelMap returns a LevelMap
-	lm, err = s.constructor()
+	var ss SeenState
+	// checks NewSeenState returns a SeenState
+	ss, err = s.constructor()
 	// and that it works
 	c.Assert(err, IsNil)
 	// setting a couple of things, sets them
-	c.Check(lm.Set("this", 12), IsNil)
-	c.Check(lm.Set("that", 42), IsNil)
-	all, err := lm.GetAll()
+	c.Check(ss.SetLevel("this", 12), IsNil)
+	c.Check(ss.SetLevel("that", 42), IsNil)
+	all, err := ss.GetAllLevels()
 	c.Check(err, IsNil)
 	c.Check(all, DeepEquals, map[string]int64{"this": 12, "that": 42})
 	// re-setting one of them, resets it
-	c.Check(lm.Set("this", 999), IsNil)
-	all, err = lm.GetAll()
+	c.Check(ss.SetLevel("this", 999), IsNil)
+	all, err = ss.GetAllLevels()
 	c.Check(err, IsNil)
 	c.Check(all, DeepEquals, map[string]int64{"this": 999, "that": 42})
 	// huzzah
 }
+
+func (s *ssSuite) TestFilterBySeen(c *C) {
+	var err error
+	var ss SeenState
+	ss, err = s.constructor()
+	// and that it works
+	c.Assert(err, IsNil)
+	n1 := protocol.Notification{MsgId: "m1"}
+	n2 := protocol.Notification{MsgId: "m2"}
+	n3 := protocol.Notification{MsgId: "m3"}
+	n4 := protocol.Notification{MsgId: "m4"}
+	n5 := protocol.Notification{MsgId: "m5"}
+
+	res, err := ss.FilterBySeen([]protocol.Notification{n1, n2, n3})
+	c.Assert(err, IsNil)
+	// everything wasn't seen yet
+	c.Check(res, DeepEquals, []protocol.Notification{n1, n2, n3})
+
+	res, err = ss.FilterBySeen([]protocol.Notification{n1, n3, n4, n5})
+	c.Assert(err, IsNil)
+	// already seen n1-n3 removed
+	c.Check(res, DeepEquals, []protocol.Notification{n4, n5})
+
+	// corner case
+	res, err = ss.FilterBySeen([]protocol.Notification{})
+	c.Assert(err, IsNil)
+	c.Assert(res, HasLen, 0)
+}

=== renamed file 'client/session/levelmap/sqlevelmap.go' => 'client/session/seenstate/sqlseenstate.go'
--- client/session/levelmap/sqlevelmap.go	2014-02-12 13:52:19 +0000
+++ client/session/seenstate/sqlseenstate.go	2014-06-05 12:03:59 +0000
@@ -14,21 +14,25 @@
  with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-package levelmap
+package seenstate
 
 import (
-	_ "code.google.com/p/gosqlite/sqlite3"
 	"database/sql"
 	"fmt"
+	"strings"
+
+	_ "code.google.com/p/gosqlite/sqlite3"
+
+	"launchpad.net/ubuntu-push/protocol"
 )
 
-type sqliteLevelMap struct {
+type sqliteSeenState struct {
 	db *sql.DB
 }
 
-// NewSqliteLevelMap returns an implementation of LevelMap that
-// persists the map in an sqlite database.
-func NewSqliteLevelMap(filename string) (LevelMap, error) {
+// NewSqliteSeenState returns an implementation of SeenState that
+// keeps and persists the state in an sqlite database.
+func NewSqliteSeenState(filename string) (SeenState, error) {
 	db, err := sql.Open("sqlite3", filename)
 	if err != nil {
 		return nil, fmt.Errorf("cannot open sqlite level map %#v: %v", filename, err)
@@ -37,18 +41,22 @@
 	if err != nil {
 		return nil, fmt.Errorf("cannot (re)create sqlite level map table: %v", err)
 	}
-	return &sqliteLevelMap{db}, nil
+	_, err = db.Exec("CREATE TABLE IF NOT EXISTS seen_msgs (id text primary key)")
+	if err != nil {
+		return nil, fmt.Errorf("cannot (re)create sqlite seen msgs table: %v", err)
+	}
+	return &sqliteSeenState{db}, nil
 }
 
-func (pm *sqliteLevelMap) Set(level string, top int64) error {
-	_, err := pm.db.Exec("REPLACE INTO level_map (level, top) VALUES (?, ?)", level, top)
+func (ps *sqliteSeenState) SetLevel(level string, top int64) error {
+	_, err := ps.db.Exec("REPLACE INTO level_map (level, top) VALUES (?, ?)", level, top)
 	if err != nil {
 		return fmt.Errorf("cannot set %#v to %#v in level map: %v", level, top, err)
 	}
 	return nil
 }
-func (pm *sqliteLevelMap) GetAll() (map[string]int64, error) {
-	rows, err := pm.db.Query("SELECT * FROM level_map")
+func (ps *sqliteSeenState) GetAllLevels() (map[string]int64, error) {
+	rows, err := ps.db.Query("SELECT * FROM level_map")
 	if err != nil {
 		return nil, fmt.Errorf("cannot retrieve levels from sqlite level map: %v", err)
 	}
@@ -64,3 +72,30 @@
 	}
 	return m, nil
 }
+
+func (ps *sqliteSeenState) dropPrevThan(msgId string) error {
+	_, err := ps.db.Exec("DELETE FROM seen_msgs WHERE rowid < (SELECT rowid FROM seen_msgs WHERE id = ?)", msgId)
+	return err
+}
+
+func (ps *sqliteSeenState) FilterBySeen(notifs []protocol.Notification) ([]protocol.Notification, error) {
+	if len(notifs) == 0 {
+		return nil, nil
+	}
+	acc := make([]protocol.Notification, 0, len(notifs))
+	for _, notif := range notifs {
+		_, err := ps.db.Exec("INSERT INTO seen_msgs (id) VALUES (?)", notif.MsgId)
+		if err != nil {
+			if strings.HasSuffix(err.Error(), "UNIQUE constraint failed: seen_msgs.id") {
+				continue
+			}
+			return nil, fmt.Errorf("cannot insert %#v in seen msgs: %v", notif.MsgId, err)
+		}
+		acc = append(acc, notif)
+	}
+	err := ps.dropPrevThan(notifs[0].MsgId)
+	if err != nil {
+		return nil, fmt.Errorf("cannot delete obsolete seen msgs: %v", err)
+	}
+	return acc, nil
+}

=== renamed file 'client/session/levelmap/sqlevelmap_test.go' => 'client/session/seenstate/sqlseenstate_test.go'
--- client/session/levelmap/sqlevelmap_test.go	2014-02-08 18:08:55 +0000
+++ client/session/seenstate/sqlseenstate_test.go	2014-06-05 12:03:59 +0000
@@ -14,29 +14,32 @@
  with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-package levelmap
+package seenstate
 
 import (
+	"database/sql"
+
 	_ "code.google.com/p/gosqlite/sqlite3"
-	"database/sql"
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/protocol"
 )
 
-type sqlmSuite struct{ lmSuite }
-
-var _ = Suite(&sqlmSuite{})
-
-func (s *sqlmSuite) SetUpSuite(c *C) {
-	s.constructor = func() (LevelMap, error) { return NewSqliteLevelMap(":memory:") }
+type sqlsSuite struct{ ssSuite }
+
+var _ = Suite(&sqlsSuite{})
+
+func (s *sqlsSuite) SetUpSuite(c *C) {
+	s.constructor = func() (SeenState, error) { return NewSqliteSeenState(":memory:") }
 }
 
-func (s *sqlmSuite) TestNewCanFail(c *C) {
-	m, err := NewSqliteLevelMap("/does/not/exist")
-	c.Assert(m, IsNil)
+func (s *sqlsSuite) TestNewCanFail(c *C) {
+	sqls, err := NewSqliteSeenState("/does/not/exist")
+	c.Assert(sqls, IsNil)
 	c.Check(err, NotNil)
 }
 
-func (s *sqlmSuite) TestSetCanFail(c *C) {
+func (s *sqlsSuite) TestSetCanFail(c *C) {
 	dir := c.MkDir()
 	filename := dir + "test.db"
 	db, err := sql.Open("sqlite3", filename)
@@ -45,14 +48,14 @@
 	_, err = db.Exec("CREATE TABLE level_map (foo)")
 	c.Assert(err, IsNil)
 	// <evil laughter>
-	m, err := NewSqliteLevelMap(filename)
+	sqls, err := NewSqliteSeenState(filename)
 	c.Check(err, IsNil)
-	c.Assert(m, NotNil)
-	err = m.Set("foo", 42)
+	c.Assert(sqls, NotNil)
+	err = sqls.SetLevel("foo", 42)
 	c.Check(err, ErrorMatches, "cannot set .*")
 }
 
-func (s *sqlmSuite) TestGetAllCanFail(c *C) {
+func (s *sqlsSuite) TestGetAllCanFail(c *C) {
 	dir := c.MkDir()
 	filename := dir + "test.db"
 	db, err := sql.Open("sqlite3", filename)
@@ -61,15 +64,15 @@
 	_, err = db.Exec("CREATE TABLE level_map AS SELECT 'what'")
 	c.Assert(err, IsNil)
 	// <evil laughter>
-	m, err := NewSqliteLevelMap(filename)
+	sqls, err := NewSqliteSeenState(filename)
 	c.Check(err, IsNil)
-	c.Assert(m, NotNil)
-	all, err := m.GetAll()
+	c.Assert(sqls, NotNil)
+	all, err := sqls.GetAllLevels()
 	c.Check(all, IsNil)
 	c.Check(err, ErrorMatches, "cannot read level .*")
 }
 
-func (s *sqlmSuite) TestGetAllCanFailDifferently(c *C) {
+func (s *sqlsSuite) TestGetAllCanFailDifferently(c *C) {
 	dir := c.MkDir()
 	filename := dir + "test.db"
 	db, err := sql.Open("sqlite3", filename)
@@ -83,10 +86,85 @@
 	_, err = db.Exec("DROP TABLE foo")
 	c.Assert(err, IsNil)
 	// <evil laughter>
-	m, err := NewSqliteLevelMap(filename)
+	sqls, err := NewSqliteSeenState(filename)
 	c.Check(err, IsNil)
-	c.Assert(m, NotNil)
-	all, err := m.GetAll()
+	c.Assert(sqls, NotNil)
+	all, err := sqls.GetAllLevels()
 	c.Check(all, IsNil)
 	c.Check(err, ErrorMatches, "cannot retrieve levels .*")
 }
+
+func (s *sqlsSuite) TestFilterBySeenCanFail(c *C) {
+	dir := c.MkDir()
+	filename := dir + "test.db"
+	db, err := sql.Open("sqlite3", filename)
+	c.Assert(err, IsNil)
+	// create the wrong kind of table
+	_, err = db.Exec("CREATE TABLE seen_msgs AS SELECT 'what'")
+	c.Assert(err, IsNil)
+	// <evil laughter>
+	sqls, err := NewSqliteSeenState(filename)
+	c.Check(err, IsNil)
+	c.Assert(sqls, NotNil)
+	n1 := protocol.Notification{MsgId: "m1"}
+	res, err := sqls.FilterBySeen([]protocol.Notification{n1})
+	c.Check(res, IsNil)
+	c.Check(err, ErrorMatches, "cannot insert .*")
+}
+
+func (s *sqlsSuite) TestDropPrevThan(c *C) {
+	dir := c.MkDir()
+	filename := dir + "test.db"
+	db, err := sql.Open("sqlite3", filename)
+	c.Assert(err, IsNil)
+	sqls, err := NewSqliteSeenState(filename)
+	c.Check(err, IsNil)
+	c.Assert(sqls, NotNil)
+
+	_, err = db.Exec("INSERT INTO seen_msgs (id) VALUES (?)", "m1")
+	c.Assert(err, IsNil)
+	_, err = db.Exec("INSERT INTO seen_msgs (id) VALUES (?)", "m2")
+	c.Assert(err, IsNil)
+	_, err = db.Exec("INSERT INTO seen_msgs (id) VALUES (?)", "m3")
+	c.Assert(err, IsNil)
+	_, err = db.Exec("INSERT INTO seen_msgs (id) VALUES (?)", "m4")
+	c.Assert(err, IsNil)
+	_, err = db.Exec("INSERT INTO seen_msgs (id) VALUES (?)", "m5")
+	c.Assert(err, IsNil)
+
+	rows, err := db.Query("SELECT COUNT(*) FROM seen_msgs")
+	c.Assert(err, IsNil)
+	rows.Next()
+	var i int
+	err = rows.Scan(&i)
+	c.Assert(err, IsNil)
+	c.Check(i, Equals, 5)
+	rows.Close()
+
+	err = sqls.(*sqliteSeenState).dropPrevThan("m3")
+	c.Assert(err, IsNil)
+
+	rows, err = db.Query("SELECT COUNT(*) FROM seen_msgs")
+	c.Assert(err, IsNil)
+	rows.Next()
+	err = rows.Scan(&i)
+	c.Assert(err, IsNil)
+	c.Check(i, Equals, 3)
+	rows.Close()
+
+	var msgId string
+	rows, err = db.Query("SELECT * FROM seen_msgs")
+	rows.Next()
+	err = rows.Scan(&msgId)
+	c.Assert(err, IsNil)
+	c.Check(msgId, Equals, "m3")
+	rows.Next()
+	err = rows.Scan(&msgId)
+	c.Assert(err, IsNil)
+	c.Check(msgId, Equals, "m4")
+	rows.Next()
+	err = rows.Scan(&msgId)
+	c.Assert(err, IsNil)
+	c.Check(msgId, Equals, "m5")
+	rows.Close()
+}

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-04-18 16:37:31 +0000
+++ client/session/session.go	2014-06-05 12:03:59 +0000
@@ -26,21 +26,24 @@
 	"fmt"
 	"math/rand"
 	"net"
+	"os/exec"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"launchpad.net/ubuntu-push/client/gethosts"
-	"launchpad.net/ubuntu-push/client/session/levelmap"
+	"launchpad.net/ubuntu-push/client/session/seenstate"
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/util"
 )
 
-var wireVersionBytes = []byte{protocol.ProtocolWireVersion}
+var (
+	wireVersionBytes = []byte{protocol.ProtocolWireVersion}
+)
 
-type Notification struct {
+type BroadcastNotification struct {
 	TopLevel int64
 	Decoded  []map[string]interface{}
 }
@@ -84,6 +87,7 @@
 	ExpectAllRepairedTime  time.Duration
 	PEM                    []byte
 	Info                   map[string]interface{}
+	AuthHelper             []string
 }
 
 // ClientSession holds a client<->server session and its configuration.
@@ -91,7 +95,7 @@
 	// configuration
 	DeviceId string
 	ClientSessionConfig
-	Levels       levelmap.LevelMap
+	SeenState    seenstate.SeenState
 	Protocolator func(net.Conn) protocol.Protocol
 	// hosts
 	getHost                hostGetter
@@ -112,9 +116,12 @@
 	pingInterval time.Duration
 	retrier      util.AutoRedialer
 	// status
-	stateP *uint32
-	ErrCh  chan error
-	MsgCh  chan *Notification
+	stateP          *uint32
+	ErrCh           chan error
+	BroadcastCh     chan *BroadcastNotification
+	NotificationsCh chan *protocol.Notification
+	// authorization
+	auth string
 	// autoredial knobs
 	shouldDelayP    *uint32
 	lastAutoRedial  time.Time
@@ -138,10 +145,10 @@
 }
 
 func NewSession(serverAddrSpec string, conf ClientSessionConfig,
-	deviceId string, levelmapFactory func() (levelmap.LevelMap, error),
+	deviceId string, seenStateFactory func() (seenstate.SeenState, error),
 	log logger.Logger) (*ClientSession, error) {
 	state := uint32(Disconnected)
-	levels, err := levelmapFactory()
+	seenState, err := seenStateFactory()
 	if err != nil {
 		return nil, err
 	}
@@ -159,7 +166,7 @@
 		DeviceId:            deviceId,
 		Log:                 log,
 		Protocolator:        protocol.NewProtocol0,
-		Levels:              levels,
+		SeenState:           seenState,
 		TLS:                 &tls.Config{},
 		stateP:              &state,
 		timeSince:           time.Since,
@@ -234,6 +241,27 @@
 	return nil
 }
 
+// addAuthorization gets the authorization blob to send to the server
+// and adds it to the session.
+func (sess *ClientSession) addAuthorization() error {
+	sess.Log.Debugf("adding authorization")
+	// using a helper, for now at least
+	if len(sess.AuthHelper) == 0 {
+		// do nothing if helper is unset or empty
+		return nil
+	}
+
+	auth, err := exec.Command(sess.AuthHelper[0], sess.AuthHelper[1:]...).Output()
+	if err != nil {
+		// For now we just log the error, as we don't want to block unauthorized users
+		sess.Log.Errorf("unable to get the authorization token from the account: %v", err)
+	} else {
+		sess.auth = strings.TrimSpace(string(auth))
+	}
+
+	return nil
+}
+
 func (sess *ClientSession) resetHosts() {
 	sess.deliveryHosts = nil
 }
@@ -343,7 +371,7 @@
 	return err
 }
 
-func (sess *ClientSession) decodeBroadcast(bcast *serverMsg) *Notification {
+func (sess *ClientSession) decodeBroadcast(bcast *serverMsg) *BroadcastNotification {
 	decoded := make([]map[string]interface{}, 0)
 	for _, p := range bcast.Payloads {
 		var v map[string]interface{}
@@ -354,7 +382,7 @@
 		}
 		decoded = append(decoded, v)
 	}
-	return &Notification{
+	return &BroadcastNotification{
 		TopLevel: bcast.TopLevel,
 		Decoded:  decoded,
 	}
@@ -362,7 +390,7 @@
 
 // handle "broadcast" messages
 func (sess *ClientSession) handleBroadcast(bcast *serverMsg) error {
-	err := sess.Levels.Set(bcast.ChanId, bcast.TopLevel)
+	err := sess.SeenState.SetLevel(bcast.ChanId, bcast.TopLevel)
 	if err != nil {
 		sess.setState(Error)
 		sess.Log.Errorf("unable to set level: %v", err)
@@ -382,15 +410,44 @@
 		bcast.ChanId, bcast.AppId, bcast.TopLevel, bcast.Payloads)
 	if bcast.ChanId == protocol.SystemChannelId {
 		// the system channel id, the only one we care about for now
-		sess.Log.Debugf("sending it over")
-		sess.MsgCh <- sess.decodeBroadcast(bcast)
-		sess.Log.Debugf("sent it over")
+		sess.Log.Debugf("sending bcast over")
+		sess.BroadcastCh <- sess.decodeBroadcast(bcast)
+		sess.Log.Debugf("sent bcast over")
 	} else {
 		sess.Log.Debugf("what is this weird channel, %#v?", bcast.ChanId)
 	}
 	return nil
 }
 
+// handle "notifications" messages
+func (sess *ClientSession) handleNotifications(ucast *serverMsg) error {
+	notifs, err := sess.SeenState.FilterBySeen(ucast.Notifications)
+	if err != nil {
+		sess.setState(Error)
+		sess.Log.Errorf("unable to record msgs seen: %v", err)
+		sess.proto.WriteMessage(protocol.AckMsg{"nak"})
+		return err
+	}
+	// the server assumes if we ack the broadcast, we've updated
+	// our state. Hence the order.
+	err = sess.proto.WriteMessage(protocol.AckMsg{"ack"})
+	if err != nil {
+		sess.setState(Error)
+		sess.Log.Errorf("unable to ack notifications: %s", err)
+		return err
+	}
+	sess.clearShouldDelay()
+	for i := range notifs {
+		notif := &notifs[i]
+		sess.Log.Debugf("unicast app:%v msg:%s payload:%s",
+			notif.AppId, notif.MsgId, notif.Payload)
+		sess.Log.Debugf("sending ucast over")
+		sess.NotificationsCh <- notif
+		sess.Log.Debugf("sent ucast over")
+	}
+	return nil
+}
+
 // handle "connbroken" messages
 func (sess *ClientSession) handleConnBroken(connBroken *serverMsg) error {
 	sess.setState(Error)
@@ -422,8 +479,15 @@
 			err = sess.handlePing()
 		case "broadcast":
 			err = sess.handleBroadcast(&recv)
+		case "notifications":
+			err = sess.handleNotifications(&recv)
 		case "connbroken":
 			err = sess.handleConnBroken(&recv)
+		case "warn":
+			// XXX: current message "warn" should be "connwarn"
+			fallthrough
+		case "connwarn":
+			sess.Log.Errorf("server sent warning: %s", recv.Reason)
 		}
 		if err != nil {
 			return err
@@ -450,17 +514,16 @@
 	}
 	proto := sess.Protocolator(conn)
 	proto.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
-	levels, err := sess.Levels.GetAll()
+	levels, err := sess.SeenState.GetAllLevels()
 	if err != nil {
 		sess.setState(Error)
 		sess.Log.Errorf("unable to start: get levels: %v", err)
 		return err
 	}
 	err = proto.WriteMessage(protocol.ConnectMsg{
-		Type:     "connect",
-		DeviceId: sess.DeviceId,
-		// xxx get the SSO Authorization string from the phone
-		Authorization: "",
+		Type:          "connect",
+		DeviceId:      sess.DeviceId,
+		Authorization: sess.auth,
 		Levels:        levels,
 		Info:          sess.Info,
 	})
@@ -495,18 +558,21 @@
 
 // run calls connect, and if it works it calls start, and if it works
 // it runs loop in a goroutine, and ships its return value over ErrCh.
-func (sess *ClientSession) run(closer func(), hostGetter, connecter, starter, looper func() error) error {
+func (sess *ClientSession) run(closer func(), authChecker, hostGetter, connecter, starter, looper func() error) error {
 	closer()
-	err := hostGetter()
-	if err != nil {
-		return err
-	}
-	err = connecter()
+	if err := authChecker(); err != nil {
+		return err
+	}
+	if err := hostGetter(); err != nil {
+		return err
+	}
+	err := connecter()
 	if err == nil {
 		err = starter()
 		if err == nil {
 			sess.ErrCh = make(chan error, 1)
-			sess.MsgCh = make(chan *Notification)
+			sess.BroadcastCh = make(chan *BroadcastNotification)
+			sess.NotificationsCh = make(chan *protocol.Notification)
 			go func() { sess.ErrCh <- looper() }()
 		}
 	}
@@ -531,7 +597,7 @@
 		// keep on trying.
 		panic("can't Dial() without a protocol constructor.")
 	}
-	return sess.run(sess.doClose, sess.getHosts, sess.connect, sess.start, sess.loop)
+	return sess.run(sess.doClose, sess.addAuthorization, sess.getHosts, sess.connect, sess.start, sess.loop)
 }
 
 func init() {

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-04-18 16:37:31 +0000
+++ client/session/session_test.go	2014-06-05 12:03:59 +0000
@@ -33,8 +33,7 @@
 	. "launchpad.net/gocheck"
 
 	"launchpad.net/ubuntu-push/client/gethosts"
-	"launchpad.net/ubuntu-push/client/session/levelmap"
-	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/client/session/seenstate"
 	"launchpad.net/ubuntu-push/protocol"
 	helpers "launchpad.net/ubuntu-push/testing"
 	"launchpad.net/ubuntu-push/testing/condition"
@@ -157,17 +156,20 @@
 	return nil
 }
 
-// brokenLevelMap is a LevelMap that always breaks
-type brokenLevelMap struct{}
+// brokenSeenState is a SeenState that always breaks
+type brokenSeenState struct{}
 
-func (*brokenLevelMap) Set(string, int64) error           { return errors.New("broken.") }
-func (*brokenLevelMap) GetAll() (map[string]int64, error) { return nil, errors.New("broken.") }
+func (*brokenSeenState) SetLevel(string, int64) error            { return errors.New("broken.") }
+func (*brokenSeenState) GetAllLevels() (map[string]int64, error) { return nil, errors.New("broken.") }
+func (*brokenSeenState) FilterBySeen([]protocol.Notification) ([]protocol.Notification, error) {
+	return nil, errors.New("broken.")
+}
 
 /////
 
 type clientSessionSuite struct {
-	log  logger.Logger
-	lvls func() (levelmap.LevelMap, error)
+	log  *helpers.TestLogger
+	lvls func() (seenstate.SeenState, error)
 }
 
 func (cs *clientSessionSuite) SetUpTest(c *C) {
@@ -175,7 +177,7 @@
 }
 
 // in-memory level map testing
-var _ = Suite(&clientSessionSuite{lvls: levelmap.NewLevelMap})
+var _ = Suite(&clientSessionSuite{lvls: seenstate.NewSeenState})
 
 // sqlite level map testing
 type clientSqlevelsSessionSuite struct{ clientSessionSuite }
@@ -183,7 +185,7 @@
 var _ = Suite(&clientSqlevelsSessionSuite{})
 
 func (cs *clientSqlevelsSessionSuite) SetUpSuite(c *C) {
-	cs.lvls = func() (levelmap.LevelMap, error) { return levelmap.NewSqliteLevelMap(":memory:") }
+	cs.lvls = func() (seenstate.SeenState, error) { return seenstate.NewSqliteSeenState(":memory:") }
 }
 
 /****************************************************************
@@ -249,8 +251,8 @@
 	c.Check(err, NotNil)
 }
 
-func (cs *clientSessionSuite) TestNewSessionBadLevelMapFails(c *C) {
-	ferr := func() (levelmap.LevelMap, error) { return nil, errors.New("Busted.") }
+func (cs *clientSessionSuite) TestNewSessionBadSeenStateFails(c *C) {
+	ferr := func() (seenstate.SeenState, error) { return nil, errors.New("Busted.") }
 	sess, err := NewSession("", dummyConf, "wah", ferr, cs.log)
 	c.Check(sess, IsNil)
 	c.Assert(err, NotNil)
@@ -347,6 +349,43 @@
 }
 
 /****************************************************************
+  addAuthorization() tests
+****************************************************************/
+
+func (cs *clientSessionSuite) TestAddAuthorizationAddsAuthorization(c *C) {
+	sess := &ClientSession{Log: cs.log}
+	sess.AuthHelper = []string{"echo", "some auth"}
+	c.Assert(sess.auth, Equals, "")
+	err := sess.addAuthorization()
+	c.Assert(err, IsNil)
+	c.Check(sess.auth, Equals, "some auth")
+}
+
+func (cs *clientSessionSuite) TestAddAuthorizationIgnoresErrors(c *C) {
+	sess := &ClientSession{Log: cs.log}
+	sess.AuthHelper = []string{"sh", "-c", "echo hello; false"}
+
+	c.Assert(sess.auth, Equals, "")
+	err := sess.addAuthorization()
+	c.Assert(err, IsNil)
+	c.Check(sess.auth, Equals, "")
+}
+
+func (cs *clientSessionSuite) TestAddAuthorizationSkipsIfUnsetOrNil(c *C) {
+	sess := &ClientSession{Log: cs.log}
+	sess.AuthHelper = nil
+	c.Assert(sess.auth, Equals, "")
+	err := sess.addAuthorization()
+	c.Assert(err, IsNil)
+	c.Check(sess.auth, Equals, "")
+
+	sess.AuthHelper = []string{}
+	err = sess.addAuthorization()
+	c.Assert(err, IsNil)
+	c.Check(sess.auth, Equals, "")
+}
+
+/****************************************************************
   startConnectionAttempt()/nextHostToTry()/started tests
 ****************************************************************/
 
@@ -601,7 +640,7 @@
 	conf := ClientSessionConfig{
 		ExchangeTimeout: time.Millisecond,
 	}
-	s.sess, err = NewSession("", conf, "wah", levelmap.NewLevelMap, helpers.NewTestLogger(c, "debug"))
+	s.sess, err = NewSession("", conf, "wah", seenstate.NewSeenState, helpers.NewTestLogger(c, "debug"))
 	c.Assert(err, IsNil)
 	s.sess.Connection = &testConn{Name: "TestHandle*"}
 	s.errCh = make(chan error, 1)
@@ -609,7 +648,8 @@
 	s.downCh = make(chan interface{}, 5)
 	s.sess.proto = &testProtocol{up: s.upCh, down: s.downCh}
 	// make the message channel buffered
-	s.sess.MsgCh = make(chan *Notification, 5)
+	s.sess.BroadcastCh = make(chan *BroadcastNotification, 5)
+	s.sess.NotificationsCh = make(chan *protocol.Notification, 5)
 }
 
 func (s *msgSuite) TestHandlePingWorks(c *C) {
@@ -668,8 +708,8 @@
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
 	c.Check(<-s.errCh, Equals, nil)
-	c.Assert(len(s.sess.MsgCh), Equals, 1)
-	c.Check(<-s.sess.MsgCh, DeepEquals, &Notification{
+	c.Assert(len(s.sess.BroadcastCh), Equals, 1)
+	c.Check(<-s.sess.BroadcastCh, DeepEquals, &BroadcastNotification{
 		TopLevel: 2,
 		Decoded: []map[string]interface{}{
 			map[string]interface{}{
@@ -681,7 +721,7 @@
 		},
 	})
 	// and finally, the session keeps track of the levels
-	levels, err := s.sess.Levels.GetAll()
+	levels, err := s.sess.SeenState.GetAllLevels()
 	c.Check(err, IsNil)
 	c.Check(levels, DeepEquals, map[string]int64{"0": 2})
 }
@@ -716,11 +756,11 @@
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
 	c.Check(<-s.errCh, IsNil)
-	c.Check(len(s.sess.MsgCh), Equals, 0)
+	c.Check(len(s.sess.BroadcastCh), Equals, 0)
 }
 
-func (s *msgSuite) TestHandleBroadcastWrongBrokenLevelmap(c *C) {
-	s.sess.Levels = &brokenLevelMap{}
+func (s *msgSuite) TestHandleBroadcastBrokenSeenState(c *C) {
+	s.sess.SeenState = &brokenSeenState{}
 	msg := serverMsg{"broadcast",
 		protocol.BroadcastMsg{
 			Type:     "broadcast",
@@ -733,8 +773,9 @@
 	s.upCh <- nil // ack ok
 	// start returns with error
 	c.Check(<-s.errCh, Not(Equals), nil)
+	c.Check(s.sess.State(), Equals, Error)
 	// no message sent out
-	c.Check(len(s.sess.MsgCh), Equals, 0)
+	c.Check(len(s.sess.BroadcastCh), Equals, 0)
 	// and nak'ed it
 	c.Check(len(s.downCh), Equals, 1)
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"nak"})
@@ -767,6 +808,118 @@
 }
 
 /****************************************************************
+  handleNotifications() tests
+****************************************************************/
+
+func (s *msgSuite) TestHandleNotificationsWorks(c *C) {
+	s.sess.setShouldDelay()
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	n2 := protocol.Notification{
+		AppId:   "app2",
+		MsgId:   "b",
+		Payload: json.RawMessage(`{"m": 2}`),
+	}
+	msg := serverMsg{"notifications",
+		protocol.BroadcastMsg{},
+		protocol.NotificationsMsg{
+			Notifications: []protocol.Notification{n1, n2},
+		}, protocol.ConnBrokenMsg{}}
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	s.upCh <- nil // ack ok
+	c.Check(<-s.errCh, Equals, nil)
+	c.Check(s.sess.ShouldDelay(), Equals, false)
+	c.Assert(len(s.sess.NotificationsCh), Equals, 2)
+	c.Check(<-s.sess.NotificationsCh, DeepEquals, &n1)
+	c.Check(<-s.sess.NotificationsCh, DeepEquals, &n2)
+}
+
+func (s *msgSuite) TestHandleNotificationsFiltersSeen(c *C) {
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	n2 := protocol.Notification{
+		AppId:   "app2",
+		MsgId:   "b",
+		Payload: json.RawMessage(`{"m": 2}`),
+	}
+	msg := serverMsg{"notifications",
+		protocol.BroadcastMsg{},
+		protocol.NotificationsMsg{
+			Notifications: []protocol.Notification{n1, n2},
+		}, protocol.ConnBrokenMsg{}}
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	s.upCh <- nil // ack ok
+	c.Check(<-s.errCh, Equals, nil)
+	c.Assert(len(s.sess.NotificationsCh), Equals, 2)
+	c.Check(<-s.sess.NotificationsCh, DeepEquals, &n1)
+	c.Check(<-s.sess.NotificationsCh, DeepEquals, &n2)
+
+	// second time they get ignored
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	s.upCh <- nil // ack ok
+	c.Check(<-s.errCh, Equals, nil)
+	c.Assert(len(s.sess.NotificationsCh), Equals, 0)
+}
+
+func (s *msgSuite) TestHandleNotificationsBadAckWrite(c *C) {
+	s.sess.setShouldDelay()
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	msg := serverMsg{"notifications",
+		protocol.BroadcastMsg{},
+		protocol.NotificationsMsg{
+			Notifications: []protocol.Notification{n1},
+		}, protocol.ConnBrokenMsg{}}
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	failure := errors.New("ACK ACK ACK")
+	s.upCh <- failure
+	c.Assert(<-s.errCh, Equals, failure)
+	c.Check(s.sess.State(), Equals, Error)
+	// didn't get to clear
+	c.Check(s.sess.ShouldDelay(), Equals, true)
+}
+
+func (s *msgSuite) TestHandleNotificationsBrokenSeenState(c *C) {
+	s.sess.setShouldDelay()
+	s.sess.SeenState = &brokenSeenState{}
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	msg := serverMsg{"notifications",
+		protocol.BroadcastMsg{},
+		protocol.NotificationsMsg{
+			Notifications: []protocol.Notification{n1},
+		}, protocol.ConnBrokenMsg{}}
+	go func() { s.errCh <- s.sess.handleNotifications(&msg) }()
+	s.upCh <- nil // ack ok
+	// start returns with error
+	c.Check(<-s.errCh, Not(Equals), nil)
+	c.Check(s.sess.State(), Equals, Error)
+	// no message sent out
+	c.Check(len(s.sess.NotificationsCh), Equals, 0)
+	// and nak'ed it
+	c.Check(len(s.downCh), Equals, 1)
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"nak"})
+	// didn't get to clear
+	c.Check(s.sess.ShouldDelay(), Equals, true)
+}
+
+/****************************************************************
   handleConnBroken() tests
 ****************************************************************/
 
@@ -861,6 +1014,26 @@
 	c.Check(<-s.errCh, Equals, failure)
 }
 
+func (s *loopSuite) TestLoopNotifications(c *C) {
+	c.Check(s.sess.State(), Equals, Running)
+
+	n1 := protocol.Notification{
+		AppId:   "app1",
+		MsgId:   "a",
+		Payload: json.RawMessage(`{"m": 1}`),
+	}
+	msg := &protocol.NotificationsMsg{
+		Type:          "notifications",
+		Notifications: []protocol.Notification{n1},
+	}
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	s.upCh <- msg
+	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
+	failure := errors.New("ack")
+	s.upCh <- failure
+	c.Check(<-s.errCh, Equals, failure)
+}
+
 func (s *loopSuite) TestLoopConnBroken(c *C) {
 	c.Check(s.sess.State(), Equals, Running)
 	broken := protocol.ConnBrokenMsg{
@@ -872,6 +1045,31 @@
 	c.Check(<-s.errCh, NotNil)
 }
 
+func (s *loopSuite) TestLoopConnWarn(c *C) {
+	warn := protocol.ConnWarnMsg{
+		Type:   "warn",
+		Reason: "XXX",
+	}
+	connwarn := protocol.ConnWarnMsg{
+		Type:   "connwarn",
+		Reason: "REASON",
+	}
+	failure := errors.New("warn")
+	log := s.sess.Log.(*helpers.TestLogger)
+
+	c.Check(s.sess.State(), Equals, Running)
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	log.ResetCapture()
+	s.upCh <- warn
+	s.upCh <- connwarn
+	s.upCh <- failure
+	c.Check(<-s.errCh, Equals, failure)
+	c.Check(log.Captured(),
+		Matches, `(?ms).* warning: XXX$.*`)
+	c.Check(log.Captured(),
+		Matches, `(?ms).* warning: REASON$`)
+}
+
 /****************************************************************
   start() tests
 ****************************************************************/
@@ -898,7 +1096,7 @@
 func (cs *clientSessionSuite) TestStartFailsIfGetLevelsFails(c *C) {
 	sess, err := NewSession("", dummyConf, "wah", cs.lvls, cs.log)
 	c.Assert(err, IsNil)
-	sess.Levels = &brokenLevelMap{}
+	sess.SeenState = &brokenSeenState{}
 	sess.Connection = &testConn{Name: "TestStartConnectMessageFails"}
 	errCh := make(chan error, 1)
 	upCh := make(chan interface{}, 5)
@@ -931,9 +1129,10 @@
 
 	c.Check(takeNext(downCh), Equals, "deadline 0")
 	c.Check(takeNext(downCh), DeepEquals, protocol.ConnectMsg{
-		Type:     "connect",
-		DeviceId: sess.DeviceId,
-		Levels:   map[string]int64{},
+		Type:          "connect",
+		DeviceId:      sess.DeviceId,
+		Levels:        map[string]int64{},
+		Authorization: "",
 	})
 	upCh <- errors.New("Overflow error in /dev/null")
 	err = <-errCh
@@ -1038,6 +1237,7 @@
 	msg, ok := takeNext(downCh).(protocol.ConnectMsg)
 	c.Check(ok, Equals, true)
 	c.Check(msg.DeviceId, Equals, "wah")
+	c.Check(msg.Authorization, Equals, "")
 	c.Check(msg.Info, DeepEquals, info)
 	upCh <- nil // no error
 	upCh <- protocol.ConnAckMsg{
@@ -1054,6 +1254,22 @@
   run() tests
 ****************************************************************/
 
+func (cs *clientSessionSuite) TestRunBailsIfAuthCheckFails(c *C) {
+	sess, err := NewSession("", dummyConf, "wah", cs.lvls, cs.log)
+	c.Assert(err, IsNil)
+	failure := errors.New("TestRunBailsIfAuthCheckFails")
+	has_closed := false
+	err = sess.run(
+		func() { has_closed = true },
+		func() error { return failure },
+		nil,
+		nil,
+		nil,
+		nil)
+	c.Check(err, Equals, failure)
+	c.Check(has_closed, Equals, true)
+}
+
 func (cs *clientSessionSuite) TestRunBailsIfHostGetterFails(c *C) {
 	sess, err := NewSession("", dummyConf, "wah", cs.lvls, cs.log)
 	c.Assert(err, IsNil)
@@ -1061,6 +1277,7 @@
 	has_closed := false
 	err = sess.run(
 		func() { has_closed = true },
+		func() error { return nil },
 		func() error { return failure },
 		nil,
 		nil,
@@ -1076,6 +1293,7 @@
 	err = sess.run(
 		func() {},
 		func() error { return nil },
+		func() error { return nil },
 		func() error { return failure },
 		nil,
 		nil)
@@ -1090,6 +1308,7 @@
 		func() {},
 		func() error { return nil },
 		func() error { return nil },
+		func() error { return nil },
 		func() error { return failure },
 		nil)
 	c.Check(err, Equals, failure)
@@ -1098,23 +1317,24 @@
 func (cs *clientSessionSuite) TestRunRunsEvenIfLoopFails(c *C) {
 	sess, err := NewSession("", dummyConf, "wah", cs.lvls, cs.log)
 	c.Assert(err, IsNil)
-	// just to make a point: until here we haven't set ErrCh & MsgCh (no
+	// just to make a point: until here we haven't set ErrCh & BroadcastCh (no
 	// biggie if this stops being true)
 	c.Check(sess.ErrCh, IsNil)
-	c.Check(sess.MsgCh, IsNil)
+	c.Check(sess.BroadcastCh, IsNil)
 	failureCh := make(chan error) // must be unbuffered
-	notf := &Notification{}
+	notf := &BroadcastNotification{}
 	err = sess.run(
 		func() {},
 		func() error { return nil },
 		func() error { return nil },
 		func() error { return nil },
-		func() error { sess.MsgCh <- notf; return <-failureCh })
+		func() error { return nil },
+		func() error { sess.BroadcastCh <- notf; return <-failureCh })
 	c.Check(err, Equals, nil)
 	// if run doesn't error it sets up the channels
 	c.Assert(sess.ErrCh, NotNil)
-	c.Assert(sess.MsgCh, NotNil)
-	c.Check(<-sess.MsgCh, Equals, notf)
+	c.Assert(sess.BroadcastCh, NotNil)
+	c.Check(<-sess.BroadcastCh, Equals, notf)
 	failure := errors.New("TestRunRunsEvenIfLoopFails")
 	failureCh <- failure
 	c.Check(<-sess.ErrCh, Equals, failure)
@@ -1317,9 +1537,9 @@
 	c.Check(takeNext(downCh), Equals, protocol.AckMsg{"ack"})
 	upCh <- nil
 	// ...get bubbled up,
-	c.Check(<-sess.MsgCh, NotNil)
+	c.Check(<-sess.BroadcastCh, NotNil)
 	// and their TopLevel remembered
-	levels, err := sess.Levels.GetAll()
+	levels, err := sess.SeenState.GetAllLevels()
 	c.Check(err, IsNil)
 	c.Check(levels, DeepEquals, map[string]int64{"0": 2})
 

=== modified file 'config/config.go'
--- config/config.go	2014-04-12 08:44:39 +0000
+++ config/config.go	2014-06-05 12:03:59 +0000
@@ -268,14 +268,16 @@
 
 // used to implement -cfg@=
 type readConfigAtVal struct {
+	path string
 	accu map[string]json.RawMessage
 }
 
 func (v *readConfigAtVal) String() string {
-	return "<config.json>"
+	return v.path
 }
 
 func (v *readConfigAtVal) Set(path string) error {
+	v.path = path
 	return readOneConfig(v.accu, path)
 }
 
@@ -292,7 +294,7 @@
 		help := destField.fld.Tag.Get("help")
 		flag.Var(&val{destField, accu}, destField.configName(), help)
 	}
-	flag.Var(&readConfigAtVal{accu}, "cfg@", "get config values from file")
+	flag.Var(&readConfigAtVal{"<config.json>", accu}, "cfg@", "get config values from file")
 	flag.Parse()
 	return nil
 }
@@ -301,17 +303,25 @@
 // command line was already parsed.
 var IgnoreParsedFlags = false
 
-// ReadFiles reads configuration from a set of files. The string
-// "<flags>" can be used as a pseudo file-path, it will consider
-// command line flags, invoking flag.Parse(). Among those the flag
-// -cfg@=FILE can be used to get further config values from FILE.
-func ReadFiles(destConfig interface{}, cfgFpaths ...string) error {
+// ReadFilesDefaults reads configuration from a set of files. The
+// string "<flags>" can be used as a pseudo file-path, it will
+// consider command line flags, invoking flag.Parse(). Among those the
+// flag -cfg@=FILE can be used to get further config values from FILE.
+// Defaults for fields can be given through a map[string]interface{}.
+func ReadFilesDefaults(destConfig interface{}, defls map[string]interface{}, cfgFpaths ...string) error {
 	destValue, err := checkDestConfig("destConfig", destConfig)
 	if err != nil {
 		return err
 	}
 	// do the parsing in two phases for better error handling
 	p1 := make(map[string]json.RawMessage)
+	for field, value := range defls {
+		b, err := json.Marshal(value)
+		if err != nil {
+			return err
+		}
+		p1[field] = json.RawMessage(b)
+	}
 	readOne := false
 	for _, cfgPath := range cfgFpaths {
 		if cfgPath == "<flags>" {
@@ -336,6 +346,13 @@
 	return fillDestConfig(destValue, p1)
 }
 
+// ReadFiles reads configuration from a set of files exactly like
+// ReadFilesDefaults but no defaults can be given making all fields
+// mandatory.
+func ReadFiles(destConfig interface{}, cfgFpaths ...string) error {
+	return ReadFilesDefaults(destConfig, nil, cfgFpaths...)
+}
+
 // CompareConfigs compares the two given configuration structures. It returns a list of differing fields or nil if the config contents are the same.
 func CompareConfig(config1, config2 interface{}) ([]string, error) {
 	v1, err := checkDestConfig("config1", config1)

=== modified file 'config/config_test.go'
--- config/config_test.go	2014-04-12 08:43:32 +0000
+++ config/config_test.go	2014-06-05 12:03:59 +0000
@@ -173,6 +173,40 @@
 	c.Check(err, NotNil)
 }
 
+type testConfig2 struct {
+	A int
+	B string
+	C []string `json:"c_list"`
+	D ConfigTimeDuration
+}
+
+func (s *configSuite) TestReadFilesDefaults(c *C) {
+	var cfg testConfig2
+	tmpDir := c.MkDir()
+	emptyCfgPath := filepath.Join(tmpDir, "e.json")
+	err := ioutil.WriteFile(emptyCfgPath, []byte("{}"), os.ModePerm)
+	c.Assert(err, IsNil)
+	err = ReadFilesDefaults(&cfg, map[string]interface{}{
+		"a":      42,
+		"b":      "foo",
+		"c_list": []string{"bar", "baz"},
+		"d":      "3s",
+	}, emptyCfgPath)
+	c.Check(err, IsNil)
+	c.Check(cfg.A, Equals, 42)
+	c.Check(cfg.B, Equals, "foo")
+	c.Check(cfg.C, DeepEquals, []string{"bar", "baz"})
+	c.Check(cfg.D.TimeDuration(), Equals, 3*time.Second)
+}
+
+func (s *configSuite) TestReadFilesDefaultsError(c *C) {
+	var cfg testConfig2
+	err := ReadFilesDefaults(&cfg, map[string]interface{}{
+		"a": make(chan int),
+	})
+	c.Assert(err, NotNil)
+}
+
 type B struct {
 	BFld int
 }
@@ -193,13 +227,6 @@
 	c.Check(a, DeepEquals, A{1, B{2}, 0})
 }
 
-type testConfig2 struct {
-	A int
-	B string
-	C []string `json:"c_list"`
-	D ConfigTimeDuration
-}
-
 func (s *configSuite) TestCompareConfig(c *C) {
 	var cfg1 = testConfig2{
 		A: 1,
@@ -304,6 +331,7 @@
 	c.Check(cfg.A, Equals, 42)
 	c.Check(cfg.B, Equals, "x")
 	c.Check(cfg.C, DeepEquals, []string{"y", "z"})
+	c.Check(flag.Lookup("cfg@").Value.String(), Equals, cfgPath)
 }
 
 func (s *configFlagsSuite) TestReadUsingFlagsHelp(c *C) {

=== modified file 'debian/changelog'
--- debian/changelog	2014-04-23 11:54:00 +0000
+++ debian/changelog	2014-06-05 12:03:59 +0000
@@ -1,3 +1,10 @@
+ubuntu-push (0.3-0.ubuntu1) UNRELEASED; urgency=medium
+
+  [ John Lenton ]
+  * New upstream release.
+
+ -- John Lenton <[e-mail]>  Tue, 15 Apr 2014 14:04:35 +0100
+
 ubuntu-push (0.2.1+14.04.20140423.1-0ubuntu1) trusty; urgency=high
 
   [ Samuele Pedroni ]

=== modified file 'debian/config.json'
--- debian/config.json	2014-04-03 21:33:58 +0000
+++ debian/config.json	2014-06-05 12:03:59 +0000
@@ -1,4 +1,5 @@
 {
+    "auth_helper": [],
     "connect_timeout": "20s",
     "exchange_timeout": "30s",
     "hosts_cache_expiry": "12h",

=== modified file 'debian/control'
--- debian/control	2014-03-25 16:26:20 +0000
+++ debian/control	2014-06-05 12:03:59 +0000
@@ -1,7 +1,7 @@
 Source: ubuntu-push
 Section: net
 Priority: optional
-X-Original-Maintainer: John Lenton <[e-mail]>
+XSBC-Original-Maintainer: John Lenton <[e-mail]>
 Maintainer: Ubuntu Developers <[e-mail]>
 Build-Depends: debhelper (>= 9),
                dh-exec (>= 0.3),
@@ -11,9 +11,14 @@
                golang-go-xdg-dev,
                golang-gocheck-dev,
                golang-gosqlite-dev,
+               golang-uuid-dev,
                libgcrypt11-dev,
                libglib2.0-dev (>= 2.31.6),
                libwhoopsie-dev,
+               libubuntuoneauth-2.0-dev,
+               libdbus-1-dev,
+               libnih-dbus-dev,
+               cmake,
 Standards-Version: 3.9.5
 Homepage: http://launchpad.net/ubuntu-push
 Vcs-Bzr: lp:ubuntu-push

=== modified file 'debian/rules'
--- debian/rules	2014-03-24 12:22:55 +0000
+++ debian/rules	2014-06-05 12:03:59 +0000
@@ -2,9 +2,13 @@
 # -*- makefile -*-
 
 export DH_GOPKG := launchpad.net/ubuntu-push
-export DEB_BUILD_OPTIONS := nostrip
 export UBUNTU_PUSH_TEST_RESOURCES_ROOT := $(CURDIR)
 
+override_dh_auto_build:
+	cd $$( find ./ -type d -regex '\./[^/]*/src/launchpad.net' -printf "%h\n" | head -n1)
+	dh_auto_build --buildsystem=golang
+	(cd signing-helper && cmake . && make)
+
 override_dh_install:
 	dh_install -Xusr/bin/cmd -Xusr/bin/dev --fail-missing
 

=== modified file 'debian/ubuntu-push-client.conf'
--- debian/ubuntu-push-client.conf	2014-04-04 12:34:25 +0000
+++ debian/ubuntu-push-client.conf	2014-06-05 12:03:59 +0000
@@ -4,3 +4,4 @@
 stop on stopped dbus
 
 exec /usr/lib/ubuntu-push-client/ubuntu-push-client
+respawn

=== modified file 'debian/ubuntu-push-client.install'
--- debian/ubuntu-push-client.install	2014-03-26 16:27:19 +0000
+++ debian/ubuntu-push-client.install	2014-06-05 12:03:59 +0000
@@ -1,4 +1,5 @@
 #!/usr/bin/dh-exec
 debian/config.json /etc/xdg/ubuntu-push-client
 debian/ubuntu-push-client.conf /usr/share/upstart/sessions
+signing-helper/signing-helper /usr/lib/ubuntu-push-client
 usr/bin/ubuntu-push => /usr/lib/ubuntu-push-client/ubuntu-push-client

=== modified file 'dependencies.tsv'
--- dependencies.tsv	2014-03-12 13:23:26 +0000
+++ dependencies.tsv	2014-06-05 12:03:59 +0000
@@ -1,4 +1,5 @@
+code.google.com/p/go-uuid	hg	7dda39b2e7d5e265014674c5af696ba4186679e9	11
 code.google.com/p/gosqlite	hg	74691fb6f83716190870cde1b658538dd4b18eb0	15
-launchpad.net/go-dbus/v1	bzr	[e-mail]-20140206110213-pbzcr6ucaz3rqmnw	125
+launchpad.net/go-dbus/v1	bzr	[e-mail]-20140530132806-hpqbkxczif6o1dpz	127
 launchpad.net/go-xdg/v0	bzr	[e-mail]-20140208094800-gubd5md7cro3mtxa	10
-launchpad.net/gocheck	bzr	[e-mail]-20140127131816-zshobk1qqme626xw	86
+launchpad.net/gocheck	bzr	[e-mail]-20140225173054-xu9zlkf9kxhvow02	87

=== added directory 'nih'
=== added directory 'nih/cnih'
=== added file 'nih/cnih/cnih.go'
--- nih/cnih/cnih.go	1970-01-01 00:00:00 +0000
+++ nih/cnih/cnih.go	2014-06-05 12:03:59 +0000
@@ -0,0 +1,28 @@
+package cnih
+
+/*
+#cgo pkg-config: dbus-1 libnih libnih-dbus
+#include <stdlib.h>
+#include <nih/alloc.h>
+#include <libnih-dbus.h>
+
+// a small wrapper because cgo doesn't handle varargs
+char *cuote (const char *id) {
+    return nih_dbus_path (NULL, "", id, NULL);
+}
+*/
+import "C"
+
+import (
+	"unsafe"
+)
+
+func Quote(s []byte) string {
+	cs := C.CString(string(s))
+	defer C.free(unsafe.Pointer(cs))
+
+	cq := C.cuote(cs)
+	defer C.nih_free(unsafe.Pointer(cq))
+
+	return C.GoString(cq)[1:]
+}

=== added file 'nih/nih.go'
--- nih/nih.go	1970-01-01 00:00:00 +0000
+++ nih/nih.go	2014-06-05 12:03:59 +0000
@@ -0,0 +1,68 @@
+/*
+ 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 nih reimplements libnih-dbus's nih_dbus_path's path element
+// quoting.
+//
+// Reimplementing libnih is a wonderful exercise that everybody should persue
+// at least thrice.
+package nih
+
+import "strconv"
+
+// Quote() takes a byte slice and quotes it á la libnih.
+func Quote(s []byte) []byte {
+	if len(s) == 0 {
+		return []byte{'_'}
+	}
+	out := make([]byte, 0, 2*len(s))
+	for _, c := range s {
+		if ('0' <= c && c <= '9') || ('a' <= c && c <= 'z') || ('A' <= c && c <= 'Z') {
+			out = append(out, c)
+		} else {
+			if c < 16 {
+				out = append(out, '_', '0')
+			} else {
+				out = append(out, '_')
+			}
+			out = strconv.AppendUint(out, uint64(c), 16)
+		}
+	}
+
+	return out
+}
+
+// Quote() takes a byte slice and undoes the damage done to it by the quoting.
+func Unquote(s []byte) []byte {
+	out := make([]byte, 0, len(s))
+
+	for i := 0; i < len(s); i++ {
+		if s[i] == '_' {
+			if len(s) < i+3 {
+				break
+			}
+			num, err := strconv.ParseUint(string(s[i+1:i+3]), 16, 8)
+			if err == nil {
+				out = append(out, byte(num))
+			}
+			i += 2
+		} else {
+			out = append(out, s[i])
+		}
+	}
+
+	return out
+}

=== added file 'nih/nih_test.go'
--- nih/nih_test.go	1970-01-01 00:00:00 +0000
+++ nih/nih_test.go	2014-06-05 12:03:59 +0000
@@ -0,0 +1,57 @@
+/*
+ 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 nih
+
+import (
+	"testing"
+
+	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/nih/cnih"
+)
+
+func TestNIH(t *testing.T) { TestingT(t) }
+
+type nihSuite struct{}
+
+var _ = Suite(&nihSuite{})
+
+func (ns *nihSuite) TestQuote(c *C) {
+	for i, s := range []struct {
+		raw    []byte
+		quoted []byte
+	}{
+		{[]byte("test"), []byte("test")},
+		{[]byte("foo/bar.baz"), []byte("foo_2fbar_2ebaz")},
+		{[]byte("test_thing"), []byte("test_5fthing")},
+		{[]byte("\x01\x0f\x10\xff"), []byte("_01_0f_10_ff")},
+		{[]byte{}, []byte{'_'}},
+	} {
+		c.Check(string(s.quoted), Equals, cnih.Quote(s.raw), Commentf("iter %d (%s)", i, string(s.quoted)))
+		c.Check(string(Quote(s.raw)), DeepEquals, string(s.quoted), Commentf("iter %d (%s)", i, string(s.quoted)))
+		c.Check(Unquote(s.quoted), DeepEquals, s.raw, Commentf("iter %d (%s)", i, string(s.quoted)))
+		c.Check(string(Quote(s.raw)), Equals, cnih.Quote(s.raw), Commentf("iter %d (%s)", i, string(s.quoted)))
+	}
+
+	// check one cnih doesn't like
+	c.Check(Quote([]byte{0}), DeepEquals, []byte("_00"))
+
+	// check we don't panic with some weird ones
+	for i, s := range []string{"foo_", "foo_a", "foo_zz"} {
+		c.Check(Unquote([]byte(s)), DeepEquals, []byte("foo"), Commentf("iter %d (%s)", i, s))
+	}
+}

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-04-04 13:54:45 +0000
+++ protocol/messages.go	2014-06-05 12:03:59 +0000
@@ -20,6 +20,7 @@
 
 import (
 	"encoding/json"
+	"fmt"
 )
 
 // System channel id using a shortened hex-encoded form for the NIL UUID.
@@ -54,6 +55,14 @@
 	Split() (done bool)
 }
 
+// OnewayMsg are messages that are not to be followed by a response,
+// after sending them the session either aborts or continues.
+type OnewayMsg interface {
+	SplittableMsg
+	// continue session after the message?
+	OnewayContinue() bool
+}
+
 // CONNBROKEN message, server side is breaking the connection for reason.
 type ConnBrokenMsg struct {
 	Type string `json:"T"`
@@ -65,11 +74,35 @@
 	return true
 }
 
+func (m *ConnBrokenMsg) OnewayContinue() bool {
+	return false
+}
+
 // CONNBROKEN reasons
 const (
 	BrokenHostMismatch = "host-mismatch"
 )
 
+// CONNWARN message, server side is warning about partial functionality
+// because reason.
+type ConnWarnMsg struct {
+	Type string `json:"T"`
+	// reason
+	Reason string
+}
+
+func (m *ConnWarnMsg) Split() bool {
+	return true
+}
+func (m *ConnWarnMsg) OnewayContinue() bool {
+	return true
+}
+
+// CONNWARN reasons
+const (
+	WarnUnauthorized = "unauthorized"
+)
+
 // PING/PONG messages
 type PingPongMsg struct {
 	Type string `json:"T"`
@@ -111,8 +144,9 @@
 }
 
 // Reset resets the splitting state if the message storage is to be
-// reused.
+// reused and sets the proper Type.
 func (b *BroadcastMsg) Reset() {
+	b.Type = "broadcast"
 	b.splitting = 0
 }
 
@@ -120,6 +154,41 @@
 type NotificationsMsg struct {
 	Type          string `json:"T"`
 	Notifications []Notification
+	splitting     int
+}
+
+// Reset resets the splitting state if the message storage is to be
+// reused and sets the proper Type.
+func (m *NotificationsMsg) Reset() {
+	m.Type = "notifications"
+	m.splitting = 0
+}
+
+func (m *NotificationsMsg) Split() bool {
+	if m.splitting != 0 {
+		m.Notifications = m.Notifications[len(m.Notifications):m.splitting]
+	}
+	notifs := m.Notifications
+	var size int
+	for i, notif := range notifs {
+		size += len(notif.Payload) + len(notif.AppId) + len(notif.MsgId) + notificationOverhead
+		if size > maxPayloadSize {
+			m.splitting = len(notifs)
+			m.Notifications = notifs[:i]
+			return false
+		}
+	}
+	return true
+}
+
+var notificationOverhead int
+
+func init() {
+	buf, err := json.Marshal(Notification{})
+	if err != nil {
+		panic(fmt.Errorf("failed to compute Notification marshal overhead: %v", err))
+	}
+	notificationOverhead = len(buf) - 4 // - 4 for the null from P(ayload)
 }
 
 // A single unicast notification
@@ -130,6 +199,19 @@
 	Payload json.RawMessage `json:"P"`
 }
 
+// ExtractPayloads gets only the payloads out of a slice of notications.
+func ExtractPayloads(notifications []Notification) []json.RawMessage {
+	n := len(notifications)
+	if n == 0 {
+		return nil
+	}
+	payloads := make([]json.RawMessage, n)
+	for i := 0; i < n; i++ {
+		payloads[i] = notifications[i].Payload
+	}
+	return payloads
+}
+
 // ACKnowledgement message
 type AckMsg struct {
 	Type string `json:"T"`

=== modified file 'protocol/messages_test.go'
--- protocol/messages_test.go	2014-04-04 13:19:10 +0000
+++ protocol/messages_test.go	2014-06-05 12:03:59 +0000
@@ -100,10 +100,100 @@
 	c.Check(b.TopLevel, Equals, int64(n))
 	c.Check(n1+n2+n3, Equals, n)
 	// reset
+	b.Type = ""
 	b.Reset()
+	c.Check(b.Type, Equals, "broadcast")
 	c.Check(b.splitting, Equals, 0)
 }
 
-func (s *messagesSuite) TestSplitConnBrokenMsg(c *C) {
-	c.Check((&ConnBrokenMsg{}).Split(), Equals, true)
+func (s *messagesSuite) TestConnBrokenMsg(c *C) {
+	m := &ConnBrokenMsg{}
+	c.Check(m.Split(), Equals, true)
+	c.Check(m.OnewayContinue(), Equals, false)
+}
+
+func (s *messagesSuite) TestConnWarnMsg(c *C) {
+	m := &ConnWarnMsg{}
+	c.Check(m.Split(), Equals, true)
+	c.Check(m.OnewayContinue(), Equals, true)
+}
+
+func (s *messagesSuite) TestExtractPayloads(c *C) {
+	c.Check(ExtractPayloads(nil), IsNil)
+	p1 := json.RawMessage(`{"a":1}`)
+	p2 := json.RawMessage(`{"b":2}`)
+	ns := []Notification{Notification{Payload: p1}, Notification{Payload: p2}}
+	c.Check(ExtractPayloads(ns), DeepEquals, []json.RawMessage{p1, p2})
+}
+
+func (s *messagesSuite) TestSplitNotificationsMsgNop(c *C) {
+	n := &NotificationsMsg{
+		Type: "notifications",
+		Notifications: []Notification{
+			Notification{"app1", "msg1", json.RawMessage(`{m:1}`)},
+			Notification{"app1", "msg1", json.RawMessage(`{m:2}`)},
+		},
+	}
+	done := n.Split()
+	c.Check(done, Equals, true)
+	c.Check(cap(n.Notifications), Equals, 2)
+	c.Check(len(n.Notifications), Equals, 2)
+}
+
+var payloadFmt2 = fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2-notificationOverhead-4-6)) // 4 = app1 6 = msg%03d
+
+func manyNotifications(c int) []Notification {
+	notifs := make([]Notification, 0, 1)
+	for i := 0; i < c; i++ {
+		notifs = append(notifs, Notification{
+			"app1",
+			fmt.Sprintf("msg%03d", i),
+			json.RawMessage(fmt.Sprintf(payloadFmt2, i)),
+		})
+	}
+	return notifs
+}
+
+func (s *messagesSuite) TestSplitNotificationsMsgMany(c *C) {
+	notifs := manyNotifications(33)
+	n := len(notifs)
+	// more interesting this way
+	c.Assert(cap(notifs), Not(Equals), n)
+	nm := &NotificationsMsg{
+		Type:          "notifications",
+		Notifications: notifs,
+	}
+	done := nm.Split()
+	c.Assert(done, Equals, false)
+	n1 := len(nm.Notifications)
+	buf, err := json.Marshal(nm)
+	c.Assert(err, IsNil)
+	c.Assert(len(buf) <= 65535, Equals, true)
+	c.Check(len(buf)+len(notifs[n1].Payload) > maxPayloadSize, Equals, true)
+	done = nm.Split()
+	c.Assert(done, Equals, true)
+	n2 := len(nm.Notifications)
+	c.Check(n1+n2, Equals, n)
+
+	notifs = manyNotifications(61)
+	n = len(notifs)
+	nm = &NotificationsMsg{
+		Type:          "notifications",
+		Notifications: notifs,
+	}
+	done = nm.Split()
+	c.Assert(done, Equals, false)
+	n1 = len(nm.Notifications)
+	done = nm.Split()
+	c.Assert(done, Equals, false)
+	n2 = len(nm.Notifications)
+	done = nm.Split()
+	c.Assert(done, Equals, true)
+	n3 := len(nm.Notifications)
+	c.Check(n1+n2+n3, Equals, n)
+	// reset
+	nm.Type = ""
+	nm.Reset()
+	c.Check(nm.Type, Equals, "notifications")
+	c.Check(nm.splitting, Equals, 0)
 }

=== modified file 'protocol/state-diag-client.gv'
--- protocol/state-diag-client.gv	2014-01-16 20:07:13 +0000
+++ protocol/state-diag-client.gv	2014-06-05 12:03:59 +0000
@@ -2,7 +2,7 @@
         label = "State diagram for client";
         size="12,6";
         rankdir=LR;
-        node [shape = doublecircle]; pingTimeout;
+        node [shape = doublecircle]; pingTimeout; connBroken;
         node [shape = circle];
         start1 -> start2 [ label = "Write wire version" ];
         start2 -> start3 [ label = "Write CONNECT" ];
@@ -13,4 +13,7 @@
         broadcast -> loop [label = "Write ACK"];
         loop -> pingTimeout [
              label = "Elapsed ping interval + exchange interval"];
+        loop -> connBroken [label = "Read CONNBROKEN"];
+        loop -> warn [label = "Read CONNWARN"];
+        warn -> loop;
 }

=== modified file 'protocol/state-diag-client.svg'
--- protocol/state-diag-client.svg	2014-01-16 19:37:57 +0000
+++ protocol/state-diag-client.svg	2014-06-05 12:03:59 +0000
@@ -4,95 +4,123 @@
 <!-- Generated by graphviz version 2.26.3 (20100126.1600)
  -->
 <!-- Title: state_diagram_client Pages: 1 -->
-<svg width="864pt" height="279pt"
- viewBox="0.00 0.00 864.00 278.89" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
-<g id="graph1" class="graph" transform="scale(0.683544 0.683544) rotate(0) translate(4 404)">
+<svg width="822pt" height="432pt"
+ viewBox="0.00 0.00 822.36 432.00" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
+<g id="graph1" class="graph" transform="scale(0.650602 0.650602) rotate(0) translate(4 660)">
 <title>state_diagram_client</title>
-<polygon fill="white" stroke="white" points="-4,5 -4,-404 1261,-404 1261,5 -4,5"/>
+<polygon fill="white" stroke="white" points="-4,5 -4,-660 1261,-660 1261,5 -4,5"/>
 <text text-anchor="middle" x="628" y="-9.4" font-family="Times Roman,serif" font-size="14.00">State diagram for client</text>
 <!-- pingTimeout -->
 <g id="node1" class="node"><title>pingTimeout</title>
-<ellipse fill="none" stroke="black" cx="1180" cy="-324" rx="72.1249" ry="72.1249"/>
-<ellipse fill="none" stroke="black" cx="1180" cy="-324" rx="76.1249" ry="76.1249"/>
-<text text-anchor="middle" x="1180" y="-320.4" font-family="Times Roman,serif" font-size="14.00">pingTimeout</text>
+<ellipse fill="none" stroke="black" cx="1180" cy="-580" rx="72.1249" ry="72.1249"/>
+<ellipse fill="none" stroke="black" cx="1180" cy="-580" rx="76.1249" ry="76.1249"/>
+<text text-anchor="middle" x="1180" y="-576.4" font-family="Times Roman,serif" font-size="14.00">pingTimeout</text>
+</g>
+<!-- connBroken -->
+<g id="node2" class="node"><title>connBroken</title>
+<ellipse fill="none" stroke="black" cx="1180" cy="-413" rx="68.8251" ry="69.2965"/>
+<ellipse fill="none" stroke="black" cx="1180" cy="-413" rx="72.7978" ry="73.2965"/>
+<text text-anchor="middle" x="1180" y="-409.4" font-family="Times Roman,serif" font-size="14.00">connBroken</text>
 </g>
 <!-- start1 -->
-<g id="node2" class="node"><title>start1</title>
-<ellipse fill="none" stroke="black" cx="42" cy="-166" rx="41.2167" ry="41.7193"/>
-<text text-anchor="middle" x="42" y="-162.4" font-family="Times Roman,serif" font-size="14.00">start1</text>
+<g id="node3" class="node"><title>start1</title>
+<ellipse fill="none" stroke="black" cx="42" cy="-231" rx="41.2167" ry="41.7193"/>
+<text text-anchor="middle" x="42" y="-227.4" font-family="Times Roman,serif" font-size="14.00">start1</text>
 </g>
 <!-- start2 -->
-<g id="node4" class="node"><title>start2</title>
-<ellipse fill="none" stroke="black" cx="292" cy="-166" rx="41.2167" ry="41.7193"/>
-<text text-anchor="middle" x="292" y="-162.4" font-family="Times Roman,serif" font-size="14.00">start2</text>
+<g id="node5" class="node"><title>start2</title>
+<ellipse fill="none" stroke="black" cx="292" cy="-231" rx="41.2167" ry="41.7193"/>
+<text text-anchor="middle" x="292" y="-227.4" font-family="Times Roman,serif" font-size="14.00">start2</text>
 </g>
 <!-- start1&#45;&gt;start2 -->
 <g id="edge2" class="edge"><title>start1&#45;&gt;start2</title>
-<path fill="none" stroke="black" d="M83.5631,-166C126.547,-166 193.757,-166 240.181,-166"/>
-<polygon fill="black" stroke="black" points="240.338,-169.5 250.338,-166 240.338,-162.5 240.338,-169.5"/>
-<text text-anchor="middle" x="167" y="-171.4" font-family="Times Roman,serif" font-size="14.00">Write wire version</text>
+<path fill="none" stroke="black" d="M83.5631,-231C126.547,-231 193.757,-231 240.181,-231"/>
+<polygon fill="black" stroke="black" points="240.338,-234.5 250.338,-231 240.338,-227.5 240.338,-234.5"/>
+<text text-anchor="middle" x="167" y="-236.4" font-family="Times Roman,serif" font-size="14.00">Write wire version</text>
 </g>
 <!-- start3 -->
-<g id="node6" class="node"><title>start3</title>
-<ellipse fill="none" stroke="black" cx="526" cy="-166" rx="41.2167" ry="41.7193"/>
-<text text-anchor="middle" x="526" y="-162.4" font-family="Times Roman,serif" font-size="14.00">start3</text>
+<g id="node7" class="node"><title>start3</title>
+<ellipse fill="none" stroke="black" cx="526" cy="-231" rx="41.2167" ry="41.7193"/>
+<text text-anchor="middle" x="526" y="-227.4" font-family="Times Roman,serif" font-size="14.00">start3</text>
 </g>
 <!-- start2&#45;&gt;start3 -->
 <g id="edge4" class="edge"><title>start2&#45;&gt;start3</title>
-<path fill="none" stroke="black" d="M333.565,-166C372.875,-166 431.992,-166 474.321,-166"/>
-<polygon fill="black" stroke="black" points="474.429,-169.5 484.429,-166 474.429,-162.5 474.429,-169.5"/>
-<text text-anchor="middle" x="409" y="-171.4" font-family="Times Roman,serif" font-size="14.00">Write CONNECT</text>
+<path fill="none" stroke="black" d="M333.565,-231C372.875,-231 431.992,-231 474.321,-231"/>
+<polygon fill="black" stroke="black" points="474.429,-234.5 484.429,-231 474.429,-227.5 474.429,-234.5"/>
+<text text-anchor="middle" x="409" y="-236.4" font-family="Times Roman,serif" font-size="14.00">Write CONNECT</text>
 </g>
 <!-- loop -->
-<g id="node8" class="node"><title>loop</title>
-<ellipse fill="none" stroke="black" cx="746" cy="-166" rx="31.8198" ry="31.8198"/>
-<text text-anchor="middle" x="746" y="-162.4" font-family="Times Roman,serif" font-size="14.00">loop</text>
+<g id="node9" class="node"><title>loop</title>
+<ellipse fill="none" stroke="black" cx="746" cy="-231" rx="31.8198" ry="31.8198"/>
+<text text-anchor="middle" x="746" y="-227.4" font-family="Times Roman,serif" font-size="14.00">loop</text>
 </g>
 <!-- start3&#45;&gt;loop -->
 <g id="edge6" class="edge"><title>start3&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M567.639,-166C606.633,-166 664.616,-166 703.793,-166"/>
-<polygon fill="black" stroke="black" points="703.818,-169.5 713.818,-166 703.818,-162.5 703.818,-169.5"/>
-<text text-anchor="middle" x="641" y="-171.4" font-family="Times Roman,serif" font-size="14.00">Read CONNACK</text>
+<path fill="none" stroke="black" d="M567.639,-231C606.633,-231 664.616,-231 703.793,-231"/>
+<polygon fill="black" stroke="black" points="703.818,-234.5 713.818,-231 703.818,-227.5 703.818,-234.5"/>
+<text text-anchor="middle" x="641" y="-236.4" font-family="Times Roman,serif" font-size="14.00">Read CONNACK</text>
 </g>
 <!-- loop&#45;&gt;pingTimeout -->
 <g id="edge16" class="edge"><title>loop&#45;&gt;pingTimeout</title>
-<path fill="none" stroke="black" d="M763.666,-192.937C772.211,-204.042 783.361,-216.128 796,-224 888.06,-281.339 1012.12,-305.973 1094,-316.443"/>
-<polygon fill="black" stroke="black" points="1093.67,-319.928 1104.02,-317.68 1094.53,-312.981 1093.67,-319.928"/>
-<text text-anchor="middle" x="941" y="-319.4" font-family="Times Roman,serif" font-size="14.00">Elapsed ping interval + exchange interval</text>
+<path fill="none" stroke="black" d="M750.211,-262.971C757.458,-313.528 773.689,-408.79 796,-434 872.806,-520.784 1006.81,-556.22 1094.46,-570.528"/>
+<polygon fill="black" stroke="black" points="1093.96,-573.992 1104.39,-572.09 1095.05,-567.078 1093.96,-573.992"/>
+<text text-anchor="middle" x="941" y="-572.4" font-family="Times Roman,serif" font-size="14.00">Elapsed ping interval + exchange interval</text>
+</g>
+<!-- loop&#45;&gt;connBroken -->
+<g id="edge18" class="edge"><title>loop&#45;&gt;connBroken</title>
+<path fill="none" stroke="black" d="M755.1,-261.824C762.755,-282.438 775.756,-308.526 796,-324 883.382,-390.791 1012.39,-408.797 1096.33,-412.948"/>
+<polygon fill="black" stroke="black" points="1096.19,-416.445 1106.33,-413.388 1096.5,-409.452 1096.19,-416.445"/>
+<text text-anchor="middle" x="941" y="-417.4" font-family="Times Roman,serif" font-size="14.00">Read CONNBROKEN</text>
 </g>
 <!-- pong -->
-<g id="node10" class="node"><title>pong</title>
-<ellipse fill="none" stroke="black" cx="1180" cy="-195" rx="34.8574" ry="35.3553"/>
-<text text-anchor="middle" x="1180" y="-191.4" font-family="Times Roman,serif" font-size="14.00">pong</text>
+<g id="node11" class="node"><title>pong</title>
+<ellipse fill="none" stroke="black" cx="1180" cy="-287" rx="34.8574" ry="35.3553"/>
+<text text-anchor="middle" x="1180" y="-283.4" font-family="Times Roman,serif" font-size="14.00">pong</text>
 </g>
 <!-- loop&#45;&gt;pong -->
 <g id="edge8" class="edge"><title>loop&#45;&gt;pong</title>
-<path fill="none" stroke="black" d="M775.392,-179.044C782.046,-181.465 789.167,-183.653 796,-185 916.362,-208.722 1062.02,-203.515 1134.48,-198.706"/>
-<polygon fill="black" stroke="black" points="1134.89,-202.186 1144.62,-198.003 1134.4,-195.203 1134.89,-202.186"/>
-<text text-anchor="middle" x="941" y="-207.4" font-family="Times Roman,serif" font-size="14.00">Read PING</text>
+<path fill="none" stroke="black" d="M768.467,-253.959C776.476,-260.698 786.005,-267.259 796,-271 911.696,-314.31 1060.9,-303.343 1134.62,-293.955"/>
+<polygon fill="black" stroke="black" points="1135.49,-297.371 1144.94,-292.588 1134.57,-290.432 1135.49,-297.371"/>
+<text text-anchor="middle" x="941" y="-307.4" font-family="Times Roman,serif" font-size="14.00">Read PING</text>
 </g>
 <!-- broadcast -->
-<g id="node12" class="node"><title>broadcast</title>
-<ellipse fill="none" stroke="black" cx="1180" cy="-84" rx="58.1882" ry="58.6899"/>
-<text text-anchor="middle" x="1180" y="-80.4" font-family="Times Roman,serif" font-size="14.00">broadcast</text>
+<g id="node13" class="node"><title>broadcast</title>
+<ellipse fill="none" stroke="black" cx="1180" cy="-176" rx="58.1882" ry="58.6899"/>
+<text text-anchor="middle" x="1180" y="-172.4" font-family="Times Roman,serif" font-size="14.00">broadcast</text>
 </g>
 <!-- loop&#45;&gt;broadcast -->
 <g id="edge10" class="edge"><title>loop&#45;&gt;broadcast</title>
-<path fill="none" stroke="black" d="M770.52,-145.1C778.217,-139.607 787.053,-134.301 796,-131 917.482,-86.1746 957.924,-122.075 1086,-103 1094.61,-101.717 1103.63,-100.165 1112.53,-98.5074"/>
-<polygon fill="black" stroke="black" points="1113.34,-101.917 1122.5,-96.5998 1112.02,-95.0419 1113.34,-101.917"/>
-<text text-anchor="middle" x="941" y="-136.4" font-family="Times Roman,serif" font-size="14.00">Read BROADCAST</text>
+<path fill="none" stroke="black" d="M775.45,-218.228C782.1,-215.791 789.205,-213.528 796,-212 922.145,-183.64 957.464,-202.973 1086,-189 1094.36,-188.091 1103.12,-187.028 1111.79,-185.909"/>
+<polygon fill="black" stroke="black" points="1112.44,-189.353 1121.9,-184.574 1111.53,-182.413 1112.44,-189.353"/>
+<text text-anchor="middle" x="941" y="-217.4" font-family="Times Roman,serif" font-size="14.00">Read BROADCAST</text>
+</g>
+<!-- warn -->
+<g id="node19" class="node"><title>warn</title>
+<ellipse fill="none" stroke="black" cx="1180" cy="-63" rx="36.7696" ry="36.7696"/>
+<text text-anchor="middle" x="1180" y="-59.4" font-family="Times Roman,serif" font-size="14.00">warn</text>
+</g>
+<!-- loop&#45;&gt;warn -->
+<g id="edge20" class="edge"><title>loop&#45;&gt;warn</title>
+<path fill="none" stroke="black" d="M753.357,-199.767C760.401,-177.027 773.396,-147.441 796,-131 901.425,-54.3166 958.242,-112.935 1086,-87 1101.84,-83.7841 1119.02,-79.6061 1134.3,-75.6396"/>
+<polygon fill="black" stroke="black" points="1135.26,-79.0068 1144.04,-73.0757 1133.48,-72.2376 1135.26,-79.0068"/>
+<text text-anchor="middle" x="941" y="-136.4" font-family="Times Roman,serif" font-size="14.00">Read CONNWARN</text>
 </g>
 <!-- pong&#45;&gt;loop -->
 <g id="edge12" class="edge"><title>pong&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M1147.19,-180.867C1129.44,-173.986 1106.92,-166.463 1086,-163 980.081,-145.465 853.051,-154.36 788.368,-160.981"/>
-<polygon fill="black" stroke="black" points="787.736,-157.528 778.16,-162.06 788.472,-164.489 787.736,-157.528"/>
-<text text-anchor="middle" x="941" y="-168.4" font-family="Times Roman,serif" font-size="14.00">Write PONG</text>
+<path fill="none" stroke="black" d="M1148.22,-271.079C1130.39,-262.942 1107.48,-253.77 1086,-249 1030.54,-236.684 866.695,-232.715 788.482,-231.502"/>
+<polygon fill="black" stroke="black" points="788.085,-227.996 778.035,-231.348 787.982,-234.995 788.085,-227.996"/>
+<text text-anchor="middle" x="941" y="-254.4" font-family="Times Roman,serif" font-size="14.00">Write PONG</text>
 </g>
 <!-- broadcast&#45;&gt;loop -->
 <g id="edge14" class="edge"><title>broadcast&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M1123.8,-67.0114C1044.83,-46.6166 899.156,-22.0001 796,-81 778.946,-90.7538 767.135,-108.842 759.293,-125.833"/>
-<polygon fill="black" stroke="black" points="756.044,-124.528 755.336,-135.099 762.482,-127.277 756.044,-124.528"/>
-<text text-anchor="middle" x="941" y="-86.4" font-family="Times Roman,serif" font-size="14.00">Write ACK</text>
+<path fill="none" stroke="black" d="M1121.7,-168.205C1028.72,-156.837 851.665,-139.849 796,-167 784,-172.853 774.037,-183.132 766.245,-193.762"/>
+<polygon fill="black" stroke="black" points="763.182,-192.043 760.465,-202.284 768.975,-195.973 763.182,-192.043"/>
+<text text-anchor="middle" x="941" y="-172.4" font-family="Times Roman,serif" font-size="14.00">Write ACK</text>
+</g>
+<!-- warn&#45;&gt;loop -->
+<g id="edge22" class="edge"><title>warn&#45;&gt;loop</title>
+<path fill="none" stroke="black" d="M1144.07,-53.3553C1070.4,-35.8873 900.397,-7.71825 796,-87 779.313,-99.6722 764.14,-151.763 754.991,-189.659"/>
+<polygon fill="black" stroke="black" points="751.574,-188.904 752.686,-199.44 758.387,-190.51 751.574,-188.904"/>
 </g>
 </g>
 </svg>

=== modified file 'protocol/state-diag-session.gv'
--- protocol/state-diag-session.gv	2014-01-16 20:07:13 +0000
+++ protocol/state-diag-session.gv	2014-06-05 12:03:59 +0000
@@ -2,6 +2,7 @@
         label = "State diagram for session";
         size="12,6";
         rankdir=LR;
+        node [shape = doublecircle]; stop;
         node [shape = circle];
         start1 -> start2 [ label = "Read wire version" ];
         start2 -> start3 [ label = "Read CONNECT" ];
@@ -17,4 +18,13 @@
         split_broadcast -> split_ack_wait [label = "Write split BROADCAST"];
         split_ack_wait -> split_broadcast [label = "Read ACK"];
         split_broadcast -> loop [label = "All split msgs written"];
+        // other
+        loop -> conn_broken [label = "Receive connbroken request"];
+        loop -> conn_warn [label = "Receive connwarn request"];
+        conn_broken -> stop [label = "Write CONNBROKEN"];
+        conn_warn -> loop [label = "Write CONNWARN"];
+        // timeouts
+        ack_wait -> stop [label = "Elapsed exhange timeout"];
+        split_ack_wait -> stop [label = "Elapsed exhange timeout"];
+        pong_wait -> stop [label = "Elapsed exhange timeout"];
 }

=== modified file 'protocol/state-diag-session.svg'
--- protocol/state-diag-session.svg	2014-01-16 19:37:57 +0000
+++ protocol/state-diag-session.svg	2014-06-05 12:03:59 +0000
@@ -4,139 +4,197 @@
 <!-- Generated by graphviz version 2.26.3 (20100126.1600)
  -->
 <!-- Title: state_diagram_session Pages: 1 -->
-<svg width="864pt" height="208pt"
- viewBox="0.00 0.00 864.00 207.94" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
-<g id="graph1" class="graph" transform="scale(0.435923 0.435923) rotate(0) translate(4 473)">
+<svg width="864pt" height="266pt"
+ viewBox="0.00 0.00 864.00 265.73" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
+<g id="graph1" class="graph" transform="scale(0.367035 0.367035) rotate(0) translate(4 720)">
 <title>state_diagram_session</title>
-<polygon fill="white" stroke="white" points="-4,5 -4,-473 1979,-473 1979,5 -4,5"/>
-<text text-anchor="middle" x="987" y="-9.4" font-family="Times Roman,serif" font-size="14.00">State diagram for session</text>
+<polygon fill="white" stroke="white" points="-4,5 -4,-720 2351,-720 2351,5 -4,5"/>
+<text text-anchor="middle" x="1173" y="-9.4" font-family="Times Roman,serif" font-size="14.00">State diagram for session</text>
+<!-- stop -->
+<g id="node1" class="node"><title>stop</title>
+<ellipse fill="none" stroke="black" cx="2309" cy="-335" rx="32.0813" ry="32.5269"/>
+<ellipse fill="none" stroke="black" cx="2309" cy="-335" rx="36.0265" ry="36.5269"/>
+<text text-anchor="middle" x="2309" y="-331.4" font-family="Times Roman,serif" font-size="14.00">stop</text>
+</g>
 <!-- start1 -->
-<g id="node1" class="node"><title>start1</title>
-<ellipse fill="none" stroke="black" cx="42" cy="-294" rx="41.2167" ry="41.7193"/>
-<text text-anchor="middle" x="42" y="-290.4" font-family="Times Roman,serif" font-size="14.00">start1</text>
+<g id="node2" class="node"><title>start1</title>
+<ellipse fill="none" stroke="black" cx="42" cy="-395" rx="41.2167" ry="41.7193"/>
+<text text-anchor="middle" x="42" y="-391.4" font-family="Times Roman,serif" font-size="14.00">start1</text>
 </g>
 <!-- start2 -->
-<g id="node3" class="node"><title>start2</title>
-<ellipse fill="none" stroke="black" cx="286" cy="-294" rx="41.2167" ry="41.7193"/>
-<text text-anchor="middle" x="286" y="-290.4" font-family="Times Roman,serif" font-size="14.00">start2</text>
+<g id="node4" class="node"><title>start2</title>
+<ellipse fill="none" stroke="black" cx="286" cy="-395" rx="41.2167" ry="41.7193"/>
+<text text-anchor="middle" x="286" y="-391.4" font-family="Times Roman,serif" font-size="14.00">start2</text>
 </g>
 <!-- start1&#45;&gt;start2 -->
 <g id="edge2" class="edge"><title>start1&#45;&gt;start2</title>
-<path fill="none" stroke="black" d="M83.6679,-294C125.213,-294 189.13,-294 233.981,-294"/>
-<polygon fill="black" stroke="black" points="234.096,-297.5 244.096,-294 234.096,-290.5 234.096,-297.5"/>
-<text text-anchor="middle" x="164" y="-299.4" font-family="Times Roman,serif" font-size="14.00">Read wire version</text>
+<path fill="none" stroke="black" d="M83.6679,-395C125.213,-395 189.13,-395 233.981,-395"/>
+<polygon fill="black" stroke="black" points="234.096,-398.5 244.096,-395 234.096,-391.5 234.096,-398.5"/>
+<text text-anchor="middle" x="164" y="-400.4" font-family="Times Roman,serif" font-size="14.00">Read wire version</text>
 </g>
 <!-- start3 -->
-<g id="node5" class="node"><title>start3</title>
-<ellipse fill="none" stroke="black" cx="516" cy="-294" rx="41.2167" ry="41.7193"/>
-<text text-anchor="middle" x="516" y="-290.4" font-family="Times Roman,serif" font-size="14.00">start3</text>
+<g id="node6" class="node"><title>start3</title>
+<ellipse fill="none" stroke="black" cx="537" cy="-395" rx="41.2167" ry="41.7193"/>
+<text text-anchor="middle" x="537" y="-391.4" font-family="Times Roman,serif" font-size="14.00">start3</text>
 </g>
 <!-- start2&#45;&gt;start3 -->
 <g id="edge4" class="edge"><title>start2&#45;&gt;start3</title>
-<path fill="none" stroke="black" d="M327.651,-294C365.959,-294 422.903,-294 464.145,-294"/>
-<polygon fill="black" stroke="black" points="464.271,-297.5 474.271,-294 464.271,-290.5 464.271,-297.5"/>
-<text text-anchor="middle" x="401" y="-299.4" font-family="Times Roman,serif" font-size="14.00">Read CONNECT</text>
+<path fill="none" stroke="black" d="M327.729,-395C370.886,-395 438.364,-395 484.973,-395"/>
+<polygon fill="black" stroke="black" points="485.171,-398.5 495.171,-395 485.171,-391.5 485.171,-398.5"/>
+<text text-anchor="middle" x="401" y="-400.4" font-family="Times Roman,serif" font-size="14.00">Read CONNECT</text>
 </g>
 <!-- loop -->
-<g id="node7" class="node"><title>loop</title>
-<ellipse fill="none" stroke="black" cx="740" cy="-294" rx="31.8198" ry="31.8198"/>
-<text text-anchor="middle" x="740" y="-290.4" font-family="Times Roman,serif" font-size="14.00">loop</text>
+<g id="node8" class="node"><title>loop</title>
+<ellipse fill="none" stroke="black" cx="790" cy="-395" rx="31.8198" ry="31.8198"/>
+<text text-anchor="middle" x="790" y="-391.4" font-family="Times Roman,serif" font-size="14.00">loop</text>
 </g>
 <!-- start3&#45;&gt;loop -->
 <g id="edge6" class="edge"><title>start3&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M557.608,-294C597.53,-294 657.517,-294 697.677,-294"/>
-<polygon fill="black" stroke="black" points="697.687,-297.5 707.687,-294 697.687,-290.5 697.687,-297.5"/>
-<text text-anchor="middle" x="633" y="-299.4" font-family="Times Roman,serif" font-size="14.00">Write CONNACK</text>
+<path fill="none" stroke="black" d="M578.778,-395C625.49,-395 700.728,-395 747.665,-395"/>
+<polygon fill="black" stroke="black" points="747.805,-398.5 757.805,-395 747.805,-391.5 747.805,-398.5"/>
+<text text-anchor="middle" x="675" y="-400.4" font-family="Times Roman,serif" font-size="14.00">Write CONNACK</text>
 </g>
 <!-- ping -->
-<g id="node9" class="node"><title>ping</title>
-<ellipse fill="none" stroke="black" cx="1063" cy="-416" rx="32.0265" ry="32.5269"/>
-<text text-anchor="middle" x="1063" y="-412.4" font-family="Times Roman,serif" font-size="14.00">ping</text>
+<g id="node10" class="node"><title>ping</title>
+<ellipse fill="none" stroke="black" cx="1135" cy="-593" rx="32.0265" ry="32.5269"/>
+<text text-anchor="middle" x="1135" y="-589.4" font-family="Times Roman,serif" font-size="14.00">ping</text>
 </g>
 <!-- loop&#45;&gt;ping -->
 <g id="edge8" class="edge"><title>loop&#45;&gt;ping</title>
-<path fill="none" stroke="black" d="M754.564,-322.853C763.046,-336.78 775.035,-352.491 790,-362 861.597,-407.491 963.396,-415.983 1020.29,-416.829"/>
-<polygon fill="black" stroke="black" points="1020.35,-420.33 1030.38,-416.906 1020.4,-413.33 1020.35,-420.33"/>
-<text text-anchor="middle" x="881" y="-418.4" font-family="Times Roman,serif" font-size="14.00">Elapsed ping interval</text>
+<path fill="none" stroke="black" d="M800.39,-425.317C809.609,-448.006 825.187,-478.237 848,-497 920.691,-556.785 1032.18,-579.907 1092.58,-588.403"/>
+<polygon fill="black" stroke="black" points="1092.15,-591.877 1102.53,-589.734 1093.08,-584.939 1092.15,-591.877"/>
+<text text-anchor="middle" x="946" y="-583.4" font-family="Times Roman,serif" font-size="14.00">Elapsed ping interval</text>
 </g>
 <!-- broadcast -->
-<g id="node11" class="node"><title>broadcast</title>
-<ellipse fill="none" stroke="black" cx="1063" cy="-200" rx="58.1882" ry="58.6899"/>
-<text text-anchor="middle" x="1063" y="-196.4" font-family="Times Roman,serif" font-size="14.00">broadcast</text>
+<g id="node12" class="node"><title>broadcast</title>
+<ellipse fill="none" stroke="black" cx="1135" cy="-281" rx="58.1882" ry="58.6899"/>
+<text text-anchor="middle" x="1135" y="-277.4" font-family="Times Roman,serif" font-size="14.00">broadcast</text>
 </g>
 <!-- loop&#45;&gt;broadcast -->
 <g id="edge10" class="edge"><title>loop&#45;&gt;broadcast</title>
-<path fill="none" stroke="black" d="M766.046,-274.934C773.498,-270.155 781.824,-265.421 790,-262 856.828,-234.035 938.382,-217.617 994.86,-208.779"/>
-<polygon fill="black" stroke="black" points="995.396,-212.238 1004.75,-207.269 994.34,-205.318 995.396,-212.238"/>
-<text text-anchor="middle" x="881" y="-267.4" font-family="Times Roman,serif" font-size="14.00">Receive broadcast request</text>
+<path fill="none" stroke="black" d="M811.332,-370.953C821.492,-360.892 834.388,-349.946 848,-343 917.32,-307.628 1006.03,-292.395 1066.35,-285.86"/>
+<polygon fill="black" stroke="black" points="1066.94,-289.319 1076.53,-284.811 1066.22,-282.355 1066.94,-289.319"/>
+<text text-anchor="middle" x="946" y="-348.4" font-family="Times Roman,serif" font-size="14.00">Receive broadcast request</text>
+</g>
+<!-- conn_broken -->
+<g id="node26" class="node"><title>conn_broken</title>
+<ellipse fill="none" stroke="black" cx="1361" cy="-99" rx="73.0388" ry="73.5391"/>
+<text text-anchor="middle" x="1361" y="-95.4" font-family="Times Roman,serif" font-size="14.00">conn_broken</text>
+</g>
+<!-- loop&#45;&gt;conn_broken -->
+<g id="edge28" class="edge"><title>loop&#45;&gt;conn_broken</title>
+<path fill="none" stroke="black" d="M793.216,-363.054C799.833,-304.219 817.014,-182.243 848,-155 967.196,-50.2026 1167.08,-63.6291 1278.91,-81.8408"/>
+<polygon fill="black" stroke="black" points="1278.34,-85.2954 1288.79,-83.4998 1279.5,-78.392 1278.34,-85.2954"/>
+<text text-anchor="middle" x="946" y="-160.4" font-family="Times Roman,serif" font-size="14.00">Receive connbroken request</text>
+</g>
+<!-- conn_warn -->
+<g id="node28" class="node"><title>conn_warn</title>
+<ellipse fill="none" stroke="black" cx="1135" cy="-477" rx="65.7609" ry="65.7609"/>
+<text text-anchor="middle" x="1135" y="-473.4" font-family="Times Roman,serif" font-size="14.00">conn_warn</text>
+</g>
+<!-- loop&#45;&gt;conn_warn -->
+<g id="edge30" class="edge"><title>loop&#45;&gt;conn_warn</title>
+<path fill="none" stroke="black" d="M814.092,-416.512C823.957,-424.185 835.89,-432.126 848,-437 915.942,-464.343 999.421,-473.523 1058.8,-476.355"/>
+<polygon fill="black" stroke="black" points="1058.71,-479.855 1068.85,-476.786 1059.01,-472.861 1058.71,-479.855"/>
+<text text-anchor="middle" x="946" y="-480.4" font-family="Times Roman,serif" font-size="14.00">Receive connwarn request</text>
 </g>
 <!-- pong_wait -->
-<g id="node13" class="node"><title>pong_wait</title>
-<ellipse fill="none" stroke="black" cx="1526" cy="-406" rx="62.9325" ry="62.9325"/>
-<text text-anchor="middle" x="1526" y="-402.4" font-family="Times Roman,serif" font-size="14.00">pong_wait</text>
+<g id="node14" class="node"><title>pong_wait</title>
+<ellipse fill="none" stroke="black" cx="537" cy="-653" rx="62.9325" ry="62.9325"/>
+<text text-anchor="middle" x="537" y="-649.4" font-family="Times Roman,serif" font-size="14.00">pong_wait</text>
 </g>
 <!-- ping&#45;&gt;pong_wait -->
 <g id="edge12" class="edge"><title>ping&#45;&gt;pong_wait</title>
-<path fill="none" stroke="black" d="M1095.56,-415.297C1169.19,-413.707 1350.04,-409.8 1452.36,-407.591"/>
-<polygon fill="black" stroke="black" points="1452.69,-411.084 1462.61,-407.369 1452.54,-404.086 1452.69,-411.084"/>
-<text text-anchor="middle" x="1289" y="-418.4" font-family="Times Roman,serif" font-size="14.00">Write PING</text>
+<path fill="none" stroke="black" d="M1103.4,-600.831C1035.81,-617.134 871.913,-654.261 732,-667 681.542,-671.594 668.481,-671.33 618,-667 615.134,-666.754 612.217,-666.46 609.275,-666.127"/>
+<polygon fill="black" stroke="black" points="609.573,-662.637 599.214,-664.858 608.697,-669.582 609.573,-662.637"/>
+<text text-anchor="middle" x="790" y="-670.4" font-family="Times Roman,serif" font-size="14.00">Write PING</text>
 </g>
 <!-- ack_wait -->
-<g id="node15" class="node"><title>ack_wait</title>
-<ellipse fill="none" stroke="black" cx="1526" cy="-269" rx="55.1543" ry="55.1543"/>
-<text text-anchor="middle" x="1526" y="-265.4" font-family="Times Roman,serif" font-size="14.00">ack_wait</text>
+<g id="node16" class="node"><title>ack_wait</title>
+<ellipse fill="none" stroke="black" cx="1598" cy="-373" rx="55.1543" ry="55.1543"/>
+<text text-anchor="middle" x="1598" y="-369.4" font-family="Times Roman,serif" font-size="14.00">ack_wait</text>
 </g>
 <!-- broadcast&#45;&gt;ack_wait -->
 <g id="edge14" class="edge"><title>broadcast&#45;&gt;ack_wait</title>
-<path fill="none" stroke="black" d="M1121.17,-208.669C1207.93,-221.599 1370.7,-245.856 1461.17,-259.339"/>
-<polygon fill="black" stroke="black" points="1460.9,-262.837 1471.3,-260.849 1461.93,-255.913 1460.9,-262.837"/>
-<text text-anchor="middle" x="1289" y="-257.4" font-family="Times Roman,serif" font-size="14.00">Write BROADCAST [fits one wire msg]</text>
+<path fill="none" stroke="black" d="M1193.25,-288.88C1264.95,-299.067 1390.23,-318.45 1496,-343 1508.9,-345.993 1522.57,-349.664 1535.58,-353.397"/>
+<polygon fill="black" stroke="black" points="1534.79,-356.813 1545.37,-356.254 1536.75,-350.093 1534.79,-356.813"/>
+<text text-anchor="middle" x="1361" y="-348.4" font-family="Times Roman,serif" font-size="14.00">Write BROADCAST [fits one wire msg]</text>
 </g>
 <!-- split_broadcast -->
-<g id="node17" class="node"><title>split_broadcast</title>
-<ellipse fill="none" stroke="black" cx="1526" cy="-110" rx="84.1457" ry="84.1457"/>
-<text text-anchor="middle" x="1526" y="-106.4" font-family="Times Roman,serif" font-size="14.00">split_broadcast</text>
+<g id="node18" class="node"><title>split_broadcast</title>
+<ellipse fill="none" stroke="black" cx="1598" cy="-216" rx="84.1457" ry="84.1457"/>
+<text text-anchor="middle" x="1598" y="-212.4" font-family="Times Roman,serif" font-size="14.00">split_broadcast</text>
 </g>
 <!-- broadcast&#45;&gt;split_broadcast -->
 <g id="edge16" class="edge"><title>broadcast&#45;&gt;split_broadcast</title>
-<path fill="none" stroke="black" d="M1120.7,-188.783C1199.06,-173.553 1340.01,-146.154 1433.29,-128.021"/>
-<polygon fill="black" stroke="black" points="1434.15,-131.421 1443.29,-126.077 1432.81,-124.549 1434.15,-131.421"/>
-<text text-anchor="middle" x="1289" y="-185.4" font-family="Times Roman,serif" font-size="14.00">BROADCAST does not fit one wire msg</text>
+<path fill="none" stroke="black" d="M1193.17,-272.834C1271.44,-261.846 1411.56,-242.174 1504.66,-229.104"/>
+<polygon fill="black" stroke="black" points="1505.23,-232.558 1514.65,-227.702 1504.26,-225.626 1505.23,-232.558"/>
+<text text-anchor="middle" x="1361" y="-272.4" font-family="Times Roman,serif" font-size="14.00">BROADCAST does not fit one wire msg</text>
+</g>
+<!-- pong_wait&#45;&gt;stop -->
+<g id="edge40" class="edge"><title>pong_wait&#45;&gt;stop</title>
+<path fill="none" stroke="black" d="M600.164,-653C651.344,-653 725.322,-653 790,-653 790,-653 790,-653 1972,-653 2131.11,-653 2245.53,-463.168 2289.33,-376.844"/>
+<polygon fill="black" stroke="black" points="2292.5,-378.343 2293.84,-367.834 2286.24,-375.212 2292.5,-378.343"/>
+<text text-anchor="middle" x="1361" y="-658.4" font-family="Times Roman,serif" font-size="14.00">Elapsed exhange timeout</text>
 </g>
 <!-- pong_wait&#45;&gt;loop -->
 <g id="edge18" class="edge"><title>pong_wait&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M1463.29,-398.59C1336,-383.27 1038.34,-346.004 790,-304 787.177,-303.523 784.269,-303.006 781.343,-302.468"/>
-<polygon fill="black" stroke="black" points="781.898,-299.011 771.42,-300.582 780.59,-305.888 781.898,-299.011"/>
-<text text-anchor="middle" x="1063" y="-359.4" font-family="Times Roman,serif" font-size="14.00">Read PONG</text>
+<path fill="none" stroke="black" d="M581.359,-607.765C632.774,-555.333 716.085,-470.376 760.273,-425.314"/>
+<polygon fill="black" stroke="black" points="762.774,-427.763 767.277,-418.172 757.776,-422.862 762.774,-427.763"/>
+<text text-anchor="middle" x="675" y="-574.4" font-family="Times Roman,serif" font-size="14.00">Read PONG</text>
+</g>
+<!-- ack_wait&#45;&gt;stop -->
+<g id="edge36" class="edge"><title>ack_wait&#45;&gt;stop</title>
+<path fill="none" stroke="black" d="M1653.02,-372.757C1765.96,-371.776 2032.03,-366.986 2254,-344 2256.89,-343.701 2259.85,-343.348 2262.84,-342.959"/>
+<polygon fill="black" stroke="black" points="2263.58,-346.389 2272.99,-341.517 2262.59,-339.459 2263.58,-346.389"/>
+<text text-anchor="middle" x="1972" y="-373.4" font-family="Times Roman,serif" font-size="14.00">Elapsed exhange timeout</text>
 </g>
 <!-- ack_wait&#45;&gt;loop -->
 <g id="edge20" class="edge"><title>ack_wait&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M1470.96,-271.898C1455.75,-272.644 1439.24,-273.404 1424,-274 1181.02,-283.507 889.323,-290.597 782.144,-293.057"/>
-<polygon fill="black" stroke="black" points="781.977,-289.56 772.059,-293.288 782.136,-296.558 781.977,-289.56"/>
-<text text-anchor="middle" x="1063" y="-292.4" font-family="Times Roman,serif" font-size="14.00">Read ACK</text>
+<path fill="none" stroke="black" d="M1542.83,-374.081C1445.66,-375.995 1237.64,-380.146 1062,-384 966.886,-386.087 942.947,-382.989 848,-389 842.829,-389.327 837.406,-389.764 832.04,-390.252"/>
+<polygon fill="black" stroke="black" points="831.485,-386.79 821.871,-391.242 832.163,-393.757 831.485,-386.79"/>
+<text text-anchor="middle" x="1135" y="-389.4" font-family="Times Roman,serif" font-size="14.00">Read ACK</text>
 </g>
 <!-- split_broadcast&#45;&gt;loop -->
 <g id="edge26" class="edge"><title>split_broadcast&#45;&gt;loop</title>
-<path fill="none" stroke="black" d="M1442.56,-99.9981C1336.06,-89.6718 1146.8,-79.5577 990,-115 894.383,-136.612 862.41,-141.921 790,-208 775.817,-220.943 764.522,-238.865 756.283,-254.999"/>
-<polygon fill="black" stroke="black" points="753.014,-253.718 751.785,-264.241 759.308,-256.781 753.014,-253.718"/>
-<text text-anchor="middle" x="1063" y="-120.4" font-family="Times Roman,serif" font-size="14.00">All split msgs written</text>
+<path fill="none" stroke="black" d="M1515.33,-201.109C1409.22,-184.681 1219.98,-164.458 1062,-196 960.985,-216.168 924.079,-215.554 848,-285 827.351,-303.849 812.751,-331.776 803.364,-354.774"/>
+<polygon fill="black" stroke="black" points="800.027,-353.699 799.658,-364.287 806.549,-356.24 800.027,-353.699"/>
+<text text-anchor="middle" x="1135" y="-201.4" font-family="Times Roman,serif" font-size="14.00">All split msgs written</text>
 </g>
 <!-- split_ack_wait -->
-<g id="node21" class="node"><title>split_ack_wait</title>
-<ellipse fill="none" stroke="black" cx="1893" cy="-110" rx="80.1095" ry="80.6102"/>
-<text text-anchor="middle" x="1893" y="-106.4" font-family="Times Roman,serif" font-size="14.00">split_ack_wait</text>
+<g id="node22" class="node"><title>split_ack_wait</title>
+<ellipse fill="none" stroke="black" cx="1972" cy="-257" rx="80.1095" ry="80.6102"/>
+<text text-anchor="middle" x="1972" y="-253.4" font-family="Times Roman,serif" font-size="14.00">split_ack_wait</text>
 </g>
 <!-- split_broadcast&#45;&gt;split_ack_wait -->
 <g id="edge22" class="edge"><title>split_broadcast&#45;&gt;split_ack_wait</title>
-<path fill="none" stroke="black" d="M1610.2,-110C1667.61,-110 1743.59,-110 1802.33,-110"/>
-<polygon fill="black" stroke="black" points="1802.35,-113.5 1812.35,-110 1802.34,-106.5 1802.35,-113.5"/>
-<text text-anchor="middle" x="1711" y="-115.4" font-family="Times Roman,serif" font-size="14.00">Write split BROADCAST</text>
+<path fill="none" stroke="black" d="M1680.55,-232.126C1687.12,-233.18 1693.66,-234.156 1700,-235 1760.22,-243.022 1828.36,-248.528 1881.38,-252.025"/>
+<polygon fill="black" stroke="black" points="1881.23,-255.523 1891.43,-252.676 1881.68,-248.537 1881.23,-255.523"/>
+<text text-anchor="middle" x="1783" y="-256.4" font-family="Times Roman,serif" font-size="14.00">Write split BROADCAST</text>
+</g>
+<!-- split_ack_wait&#45;&gt;stop -->
+<g id="edge38" class="edge"><title>split_ack_wait&#45;&gt;stop</title>
+<path fill="none" stroke="black" d="M2050.59,-275.189C2116.55,-290.456 2208.51,-311.74 2263.08,-324.372"/>
+<polygon fill="black" stroke="black" points="2262.62,-327.857 2273.15,-326.702 2264.2,-321.037 2262.62,-327.857"/>
+<text text-anchor="middle" x="2166" y="-327.4" font-family="Times Roman,serif" font-size="14.00">Elapsed exhange timeout</text>
 </g>
 <!-- split_ack_wait&#45;&gt;split_broadcast -->
 <g id="edge24" class="edge"><title>split_ack_wait&#45;&gt;split_broadcast</title>
-<path fill="none" stroke="black" d="M1814.64,-90.9448C1807.69,-89.7544 1800.74,-88.7397 1794,-88 1720.66,-79.9496 1701.36,-80.1783 1628,-88 1624.71,-88.3505 1621.38,-88.7628 1618.01,-89.2264"/>
-<polygon fill="black" stroke="black" points="1617.23,-85.8043 1607.87,-90.7602 1618.28,-92.7256 1617.23,-85.8043"/>
-<text text-anchor="middle" x="1711" y="-93.4" font-family="Times Roman,serif" font-size="14.00">Read ACK</text>
+<path fill="none" stroke="black" d="M1899.33,-222.493C1888.37,-218.573 1877.04,-215.2 1866,-213 1808.89,-201.617 1743.56,-202.081 1691.71,-205.529"/>
+<polygon fill="black" stroke="black" points="1691.25,-202.053 1681.52,-206.256 1691.74,-209.035 1691.25,-202.053"/>
+<text text-anchor="middle" x="1783" y="-218.4" font-family="Times Roman,serif" font-size="14.00">Read ACK</text>
+</g>
+<!-- conn_broken&#45;&gt;stop -->
+<g id="edge32" class="edge"><title>conn_broken&#45;&gt;stop</title>
+<path fill="none" stroke="black" d="M1434.61,-95.8477C1564.45,-92.5456 1841.16,-95.6985 2060,-168 2154.32,-199.163 2175.34,-218.326 2254,-279 2262.17,-285.305 2270.33,-292.782 2277.75,-300.185"/>
+<polygon fill="black" stroke="black" points="2275.42,-302.808 2284.91,-307.517 2280.43,-297.917 2275.42,-302.808"/>
+<text text-anchor="middle" x="1783" y="-127.4" font-family="Times Roman,serif" font-size="14.00">Write CONNBROKEN</text>
+</g>
+<!-- conn_warn&#45;&gt;loop -->
+<g id="edge34" class="edge"><title>conn_warn&#45;&gt;loop</title>
+<path fill="none" stroke="black" d="M1083.63,-435.301C1071.29,-427.246 1057.71,-419.822 1044,-415 972.758,-389.933 883.562,-389.406 832.05,-391.836"/>
+<polygon fill="black" stroke="black" points="831.758,-388.346 821.959,-392.373 832.131,-395.337 831.758,-388.346"/>
+<text text-anchor="middle" x="946" y="-420.4" font-family="Times Roman,serif" font-size="14.00">Write CONNWARN</text>
 </g>
 </g>
 </svg>

=== added file 'scripts/broadcast'
--- scripts/broadcast	1970-01-01 00:00:00 +0000
+++ scripts/broadcast	2014-06-05 12:03:59 +0000
@@ -0,0 +1,59 @@
+#!/usr/bin/python
+"""
+send broadcast to channel with payload data
+"""
+import argparse
+import json
+import requests
+import requests.auth
+import datetime
+import sys
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('channel', nargs=1)
+    parser.add_argument('data', nargs=1)
+    parser.add_argument('-H', '--host',
+                        help="host:port (default: %(default)s)",
+                        default="localhost:8080")
+    parser.add_argument('-e', '--expire',
+                        help="expire after the given amount of time, "
+                        "use 'd' suffix for days, 's' for seconds"
+                        " (default: %(default)s)", default="1d")
+    parser.add_argument('--no-https', action='store_true', default=False)
+    parser.add_argument('--insecure', action='store_true', default=False,
+                         help="don't check host/certs with https")
+    parser.add_argument('-u', '--user',  default="")
+    parser.add_argument('-p', '--password', default="")
+    args = parser.parse_args()
+    expire_on = datetime.datetime.utcnow()
+    ex = args.expire
+    if ex.endswith('d'):
+        delta = datetime.timedelta(days=int(ex[:-1]))
+    elif ex.endswith('s'):
+        delta = datetime.timedelta(seconds=int(ex[:-1]))
+    else:
+        print >>sys.stderr, "unknown --expire suffix:", ex
+        sys.exit(1)
+    expire_on += delta
+    scheme = 'https'
+    if args.no_https:
+        scheme = 'http'
+    url = "%s://%s/broadcast" % (scheme, args.host)
+    body = {
+        'channel': args.channel[0],
+        'data': json.loads(args.data[0]),
+        'expire_on': expire_on.replace(microsecond=0).isoformat()+"Z"
+        }
+    xauth = {}
+    if args.user and args.password:
+        xauth = {'auth': requests.auth.HTTPBasicAuth(args.user, args.password)}
+    headers = {'Content-Type': 'application/json'}
+    r = requests.post(url, data=json.dumps(body), headers=headers,
+                      verify=not args.insecure, **xauth)
+    print r.status_code
+    print r.text
+
+if __name__ == '__main__':
+    main()

=== added file 'scripts/deps.sh'
--- scripts/deps.sh	1970-01-01 00:00:00 +0000
+++ scripts/deps.sh	2014-06-05 12:03:59 +0000
@@ -0,0 +1,28 @@
+#!/bin/sh
+set -eu
+
+PROJECT=launchpad.net/ubuntu-push
+
+mktpl () {
+    for f in GoFiles CgoFiles; do
+        echo '{{join .'$f' "\\n"}}'
+    done
+}
+
+directs () {
+    go list -f "$(mktpl)" $1 | sed -e "s|^|$1|"
+}
+
+indirects () {
+    for i in $(go list -f '{{join .Deps "\n"}}' $1 | grep ^$PROJECT ); do
+        directs $i/
+    done
+    wait
+}
+
+norm () {
+    tr "\n" " " | sed -r -e "s|$PROJECT/?||g" -e 's/ *$//'
+}
+
+out="$1.deps"
+( echo -n "${1%.go} ${out}: "; indirects $(echo $1 | norm) | norm ) > "$out"

=== added file 'scripts/register'
--- scripts/register	1970-01-01 00:00:00 +0000
+++ scripts/register	2014-06-05 12:03:59 +0000
@@ -0,0 +1,43 @@
+#!/usr/bin/python
+"""
+request a unicast registration
+"""
+import argparse
+import json
+import requests
+import subprocess
+import datetime
+import sys
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('deviceid', nargs=1)
+    parser.add_argument('appid', nargs=1)
+    parser.add_argument('-H', '--host',
+                        help="host:port (default: %(default)s)",
+                        default="localhost:8080")
+    parser.add_argument('--no-https', action='store_true', default=False)
+    parser.add_argument('--insecure', action='store_true', default=False,
+                         help="don't check host/certs with https")
+    parser.add_argument('--auth_helper',  default="")
+    args = parser.parse_args()
+    scheme = 'https'
+    if args.no_https:
+        scheme = 'http'
+    url = "%s://%s/register" % (scheme, args.host)
+    body = {
+        'deviceid': args.deviceid[0],
+        'appid': args.appid[0],
+        }
+    headers = {'Content-Type': 'application/json'}
+    if args.auth_helper:
+        auth = subprocess.check_output([args.auth_helper, url]).strip()
+        headers['Authorization'] = auth
+    r = requests.post(url, data=json.dumps(body), headers=headers,
+                      verify=not args.insecure)
+    print r.status_code
+    print r.text
+
+if __name__ == '__main__':
+    main()

=== added file 'scripts/unicast'
--- scripts/unicast	1970-01-01 00:00:00 +0000
+++ scripts/unicast	2014-06-05 12:03:59 +0000
@@ -0,0 +1,67 @@
+#!/usr/bin/python
+"""
+send unicast to reg with payload data
+"""
+import argparse
+import json
+import requests
+import requests.auth
+import datetime
+import sys
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('reg', nargs=1) # userid:deviceid or reg
+    parser.add_argument('appid', nargs=1)
+    parser.add_argument('data', nargs=1)
+    parser.add_argument('-H', '--host',
+                        help="host:port (default: %(default)s)",
+                        default="localhost:8080")
+    parser.add_argument('-e', '--expire',
+                        help="expire after the given amount of time, "
+                        "use 'd' suffix for days, 's' for seconds"
+                        " (default: %(default)s)", default="1d")
+    parser.add_argument('--no-https', action='store_true', default=False)
+    parser.add_argument('--insecure', action='store_true', default=False,
+                         help="don't check host/certs with https")
+    parser.add_argument('-u', '--user',  default="")
+    parser.add_argument('-p', '--password', default="")
+    args = parser.parse_args()
+    expire_on = datetime.datetime.utcnow()
+    ex = args.expire
+    if ex.endswith('d'):
+        delta = datetime.timedelta(days=int(ex[:-1]))
+    elif ex.endswith('s'):
+        delta = datetime.timedelta(seconds=int(ex[:-1]))
+    else:
+        print >>sys.stderr, "unknown --expire suffix:", ex
+        sys.exit(1)
+    expire_on += delta
+    scheme = 'https'
+    if args.no_https:
+        scheme = 'http'
+    url = "%s://%s/notify" % (scheme, args.host)
+    body = {
+        'appid': args.appid[0],
+        'data': json.loads(args.data[0]),
+        'expire_on': expire_on.replace(microsecond=0).isoformat()+"Z"
+        }
+    reg = args.reg[0]
+    if ':' in reg:
+        userid, devid = reg.split(':', 1)
+        body['userid'] = userid
+        body['deviceid'] = devid
+    else:
+        body['token'] = reg
+    xauth = {}
+    if args.user and args.password:
+        xauth = {'auth': requests.auth.HTTPBasicAuth(args.user, args.password)}
+    headers = {'Content-Type': 'application/json'}
+    r = requests.post(url, data=json.dumps(body), headers=headers,
+                      verify=not args.insecure, **xauth)
+    print r.status_code
+    print r.text
+
+if __name__ == '__main__':
+    main()

=== modified file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	2014-04-07 19:39:19 +0000
+++ server/acceptance/acceptance_test.go	2014-06-05 12:03:59 +0000
@@ -59,3 +59,6 @@
 
 // broadcast
 var _ = Suite(&suites.BroadcastAcceptanceSuite{suites.AcceptanceSuite{StartServer: StartServer}})
+
+// unicast
+var _ = Suite(&suites.UnicastAcceptanceSuite{suites.AcceptanceSuite{StartServer: StartServer}, nil})

=== modified file 'server/acceptance/acceptanceclient.go'
--- server/acceptance/acceptanceclient.go	2014-04-09 19:30:53 +0000
+++ server/acceptance/acceptanceclient.go	2014-06-05 12:03:59 +0000
@@ -24,6 +24,7 @@
 	"errors"
 	"fmt"
 	"net"
+	"strings"
 	"time"
 
 	"launchpad.net/ubuntu-push/protocol"
@@ -44,6 +45,7 @@
 	Levels          map[string]int64
 	Insecure        bool   // don't verify certs
 	Prefix          string // prefix for events
+	Auth            string
 	// connection
 	Connection net.Conn
 }
@@ -73,6 +75,7 @@
 	Type string `json:"T"`
 	protocol.BroadcastMsg
 	protocol.NotificationsMsg
+	protocol.ConnWarnMsg
 }
 
 // Run the session with the server, emits a stream of events.
@@ -93,6 +96,7 @@
 			"device":  sess.Model,
 			"channel": sess.ImageChannel,
 		},
+		Authorization: sess.Auth,
 	})
 	if err != nil {
 		return err
@@ -125,9 +129,24 @@
 			if sess.ReportPings {
 				events <- sess.Prefix + "ping"
 			}
+		case "notifications":
+			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+			err := proto.WriteMessage(protocol.AckMsg{Type: "ack"})
+			if err != nil {
+				return err
+			}
+			parts := make([]string, len(recv.Notifications))
+			for i, notif := range recv.Notifications {
+				pack, err := json.Marshal(&notif.Payload)
+				if err != nil {
+					return err
+				}
+				parts[i] = fmt.Sprintf("app:%v payload:%s;", notif.AppId, pack)
+			}
+			events <- fmt.Sprintf("%sunicast %s", sess.Prefix, strings.Join(parts, " "))
 		case "broadcast":
 			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
-			err := proto.WriteMessage(protocol.PingPongMsg{Type: "ack"})
+			err := proto.WriteMessage(protocol.AckMsg{Type: "ack"})
 			if err != nil {
 				return err
 			}
@@ -136,6 +155,8 @@
 				return err
 			}
 			events <- fmt.Sprintf("%sbroadcast chan:%v app:%v topLevel:%d payloads:%s", sess.Prefix, recv.ChanId, recv.AppId, recv.TopLevel, pack)
+		case "warn", "connwarn":
+			events <- fmt.Sprintf("%sconnwarn %s", sess.Prefix, recv.Reason)
 		}
 	}
 	return nil

=== modified file 'server/acceptance/cmd/acceptanceclient.go'
--- server/acceptance/cmd/acceptanceclient.go	2014-04-14 14:54:14 +0000
+++ server/acceptance/cmd/acceptanceclient.go	2014-06-05 12:03:59 +0000
@@ -22,7 +22,9 @@
 	"fmt"
 	"log"
 	"os"
+	"os/exec"
 	"path/filepath"
+	"strings"
 
 	"launchpad.net/ubuntu-push/config"
 	"launchpad.net/ubuntu-push/server/acceptance"
@@ -40,12 +42,13 @@
 	ExchangeTimeout config.ConfigTimeDuration `json:"exchange_timeout"`
 	// server connection config
 	Addr        config.ConfigHostPort
-	CertPEMFile string `json:"cert_pem_file"`
+	CertPEMFile string   `json:"cert_pem_file"`
+	AuthHelper  []string `json:"auth_helper"`
 }
 
 func main() {
 	flag.Usage = func() {
-		fmt.Fprintf(os.Stderr, "Usage: acceptancclient [options] <config.json> <device id>\n")
+		fmt.Fprintf(os.Stderr, "Usage: acceptancclient [options] <device id>\n")
 		flag.PrintDefaults()
 	}
 	missingArg := func(what string) {
@@ -53,28 +56,24 @@
 		flag.Usage()
 		os.Exit(2)
 	}
-	flag.Parse()
+	cfg := &configuration{}
+	err := config.ReadFilesDefaults(cfg, map[string]interface{}{
+		"exchange_timeout": "5s",
+		"cert_pem_file":    "",
+		"auth_helper":      []string{},
+	}, "<flags>")
+	if err != nil {
+		log.Fatalf("reading config: %v", err)
+	}
 	narg := flag.NArg()
 	switch {
 	case narg < 1:
-		missingArg("config file")
-	case narg < 2:
 		missingArg("device-id")
 	}
-	configFName := flag.Arg(0)
-	f, err := os.Open(configFName)
-	if err != nil {
-		log.Fatalf("reading config: %v", err)
-	}
-	cfg := &configuration{}
-	err = config.ReadConfig(f, cfg)
-	if err != nil {
-		log.Fatalf("reading config: %v", err)
-	}
 	session := &acceptance.ClientSession{
 		ExchangeTimeout: cfg.ExchangeTimeout.TimeDuration(),
 		ServerAddr:      cfg.Addr.HostPort(),
-		DeviceId:        flag.Arg(1),
+		DeviceId:        flag.Arg(0),
 		// flags
 		Model:        *deviceModel,
 		ImageChannel: *imageChannel,
@@ -82,12 +81,23 @@
 		Insecure:     *insecureFlag,
 	}
 	log.Printf("with: %#v", session)
-	if !*insecureFlag {
-		session.CertPEMBlock, err = config.LoadFile(cfg.CertPEMFile, filepath.Dir(configFName))
+	if !*insecureFlag && cfg.CertPEMFile != "" {
+		cfgDir := filepath.Dir(flag.Lookup("cfg@").Value.String())
+		log.Printf("cert: %v relToDir: %v", cfg.CertPEMFile, cfgDir)
+		session.CertPEMBlock, err = config.LoadFile(cfg.CertPEMFile, cfgDir)
 		if err != nil {
 			log.Fatalf("reading CertPEMFile: %v", err)
 		}
 	}
+	if len(cfg.AuthHelper) != 0 {
+		helperArgs := cfg.AuthHelper[1:]
+		helperArgs = append(helperArgs, "https://push.ubuntu.com/")
+		auth, err := exec.Command(cfg.AuthHelper[0], helperArgs...).Output()
+		if err != nil {
+			log.Fatalf("auth helper: %v", err)
+		}
+		session.Auth = strings.TrimSpace(string(auth))
+	}
 	err = session.Dial()
 	if err != nil {
 		log.Fatalln(err)

=== modified file 'server/acceptance/suites/broadcast.go'
--- server/acceptance/suites/broadcast.go	2014-04-07 19:39:19 +0000
+++ server/acceptance/suites/broadcast.go	2014-06-05 12:03:59 +0000
@@ -29,14 +29,11 @@
 	"launchpad.net/ubuntu-push/server/api"
 )
 
-// BroadCastAcceptanceSuite has tests about broadcast.
+// BroadcastAcceptanceSuite has tests about broadcast.
 type BroadcastAcceptanceSuite struct {
 	AcceptanceSuite
 }
 
-// Long after the end of the tests.
-var future = time.Now().Add(9 * time.Hour).Format(time.RFC3339)
-
 func (s *BroadcastAcceptanceSuite) TestBroadcastToConnected(c *C) {
 	events, errCh, stop := s.StartClient(c, "DEVB", nil)
 	got, err := s.PostRequest("/broadcast", &api.Broadcast{
@@ -91,7 +88,7 @@
 	c.Check(len(errCh), Equals, 0)
 }
 
-func (s *BroadcastAcceptanceSuite) TestBroadcasLargeNeedsSplitting(c *C) {
+func (s *BroadcastAcceptanceSuite) TestBroadcastLargeNeedsSplitting(c *C) {
 	// send bunch of broadcasts that will be pending
 	payloadFmt := fmt.Sprintf(`{"img1/m1":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
 	for i := 0; i < 32; i++ {
@@ -106,8 +103,17 @@
 
 	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:\[{"img1/m1":0,.*`)
-	c.Check(NextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:32 payloads:\[.*`)
+	n := 0
+	for {
+		evt := NextEvent(events, errCh)
+		c.Check(evt, Matches, "broadcast chan:0 .*")
+		n += 1
+		if strings.Contains(evt, "topLevel:32") {
+			break
+		}
+	}
+	// was split
+	c.Check(n > 1, Equals, true)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
@@ -265,7 +271,11 @@
 
 func (s *BroadcastAcceptanceSuite) TestGetHosts(c *C) {
 	gh := gethosts.New("", s.ServerAPIURL+"/delivery-hosts", 2*time.Second)
-	hosts, err := gh.Get()
+	host, err := gh.Get()
 	c.Assert(err, IsNil)
-	c.Check(hosts, DeepEquals, []string{s.ServerAddr})
+	expected := &gethosts.Host{
+		Domain: "localhost",
+		Hosts:  []string{s.ServerAddr},
+	}
+	c.Check(host, DeepEquals, expected)
 }

=== modified file 'server/acceptance/suites/suite.go'
--- server/acceptance/suites/suite.go	2014-04-03 16:47:47 +0000
+++ server/acceptance/suites/suite.go	2014-06-05 12:03:59 +0000
@@ -44,10 +44,19 @@
 
 // Start a client.
 func (h *ServerHandle) StartClient(c *C, devId string, levels map[string]int64) (events <-chan string, errorCh <-chan error, stop func()) {
+	return h.StartClientAuth(c, devId, levels, "")
+}
+
+// Start a client with auth.
+func (h *ServerHandle) StartClientAuth(c *C, devId string, levels map[string]int64, auth string) (events <-chan string, errorCh <-chan error, stop func()) {
 	errCh := make(chan error, 1)
 	cliEvents := make(chan string, 10)
 	sess := testClientSession(h.ServerAddr, devId, "m1", "img1", false)
 	sess.Levels = levels
+	sess.Auth = auth
+	if auth != "" {
+		sess.ExchangeTimeout = 5 * time.Second
+	}
 	err := sess.Dial()
 	c.Assert(err, IsNil)
 	clientShutdown := make(chan bool, 1) // abused as an atomic flag
@@ -80,7 +89,7 @@
 	// to kill the server process
 	KillGroup map[string]func(os.Signal)
 	// hook to adjust requests
-	MassageRequest func(req *http.Request) *http.Request
+	MassageRequest func(req *http.Request, message interface{}) *http.Request
 	// other state
 	httpClient *http.Client
 }
@@ -115,7 +124,7 @@
 	request.Header.Set("Content-Type", "application/json")
 
 	if s.MassageRequest != nil {
-		request = s.MassageRequest(request)
+		request = s.MassageRequest(request, message)
 	}
 
 	resp, err := s.httpClient.Do(request)
@@ -186,3 +195,6 @@
 	}
 	return
 }
+
+// Long after the end of the tests.
+var future = time.Now().Add(9 * time.Hour).Format(time.RFC3339)

=== added file 'server/acceptance/suites/unicast.go'
--- server/acceptance/suites/unicast.go	1970-01-01 00:00:00 +0000
+++ server/acceptance/suites/unicast.go	2014-06-05 12:03:59 +0000
@@ -0,0 +1,157 @@
+/*
+ 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 suites
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/server/api"
+)
+
+// UnicastAcceptanceSuite has tests about unicast.
+type UnicastAcceptanceSuite struct {
+	AcceptanceSuite
+	AssociatedAuth func(string) (string, string)
+}
+
+func (s *UnicastAcceptanceSuite) associatedAuth(deviceId string) (userId string, auth string) {
+	if s.AssociatedAuth != nil {
+		return s.AssociatedAuth(deviceId)
+	}
+	return deviceId, ""
+}
+
+func (s *UnicastAcceptanceSuite) TestUnicastToConnected(c *C) {
+	_, auth := s.associatedAuth("DEV1")
+	res, err := s.PostRequest("/register", &api.Registration{
+		DeviceId: "DEV1",
+		AppId:    "app1",
+	})
+	c.Assert(err, IsNil)
+	c.Assert(res, Matches, ".*ok.*")
+	var reg map[string]interface{}
+	err = json.Unmarshal([]byte(res), &reg)
+	c.Assert(err, IsNil)
+	events, errCh, stop := s.StartClientAuth(c, "DEV1", nil, auth)
+	got, err := s.PostRequest("/notify", &api.Unicast{
+		Token:    reg["token"].(string),
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 42}`),
+	})
+	c.Assert(err, IsNil)
+	c.Assert(got, Matches, ".*ok.*")
+	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)
+}
+
+func (s *UnicastAcceptanceSuite) TestUnicastCorrectDistribution(c *C) {
+	userId1, auth1 := s.associatedAuth("DEV1")
+	userId2, auth2 := s.associatedAuth("DEV2")
+	// start 1st client
+	events1, errCh1, stop1 := s.StartClientAuth(c, "DEV1", nil, auth1)
+	// start 2nd client
+	events2, errCh2, stop2 := s.StartClientAuth(c, "DEV2", nil, auth2)
+	// unicast to one and the other
+	got, err := s.PostRequest("/notify", &api.Unicast{
+		UserId:   userId1,
+		DeviceId: "DEV1",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"to": 1}`),
+	})
+	c.Assert(err, IsNil)
+	c.Assert(got, Matches, ".*ok.*")
+	got, err = s.PostRequest("/notify", &api.Unicast{
+		UserId:   userId2,
+		DeviceId: "DEV2",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"to": 2}`),
+	})
+	c.Assert(err, IsNil)
+	c.Assert(got, Matches, ".*ok.*")
+	c.Check(NextEvent(events1, errCh1), Equals, `unicast app:app1 payload:{"to":1};`)
+	c.Check(NextEvent(events2, errCh2), Equals, `unicast app:app1 payload:{"to":2};`)
+	stop1()
+	stop2()
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
+	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)
+}
+
+func (s *UnicastAcceptanceSuite) TestUnicastLargeNeedsSplitting(c *C) {
+	userId, auth := s.associatedAuth("DEV2")
+	// send bunch of unicasts that will be pending
+	payloadFmt := fmt.Sprintf(`{"serial":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
+	for i := 0; i < 32; i++ {
+		got, err := s.PostRequest("/notify", &api.Unicast{
+			UserId:   userId,
+			DeviceId: "DEV2",
+			AppId:    "app1",
+			ExpireOn: future,
+			Data:     json.RawMessage(fmt.Sprintf(payloadFmt, i)),
+		})
+		c.Assert(err, IsNil)
+		c.Assert(got, Matches, ".*ok.*")
+	}
+
+	events, errCh, stop := s.StartClientAuth(c, "DEV2", nil, auth)
+	// gettting pending on connect
+	n := 0
+	for {
+		evt := NextEvent(events, errCh)
+		c.Check(evt, Matches, "unicast app:app1 .*")
+		n += 1
+		if strings.Contains(evt, `"serial":31`) {
+			break
+		}
+	}
+	// was split
+	c.Check(n > 1, Equals, true)
+	stop()
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}

=== modified file 'server/api/handlers.go'
--- server/api/handlers.go	2014-02-20 17:09:03 +0000
+++ server/api/handlers.go	2014-06-05 12:03:59 +0000
@@ -19,12 +19,15 @@
 package api
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"time"
 
+	"code.google.com/p/go-uuid/uuid"
+
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/store"
@@ -48,6 +51,8 @@
 	ioError        = "io-error"
 	invalidRequest = "invalid-request"