Merge lp:~chipaca/ubuntu-push/endpoint-names into lp:ubuntu-push

[John Lenton]

Added endpoint.GrabName

Added endpoint.GrabName

Unmerged revisions

=== modified file 'Makefile'
--- Makefile	2014-03-31 17:58:54 +0000
+++ Makefile	2014-05-12 14:11:46 +0000
@@ -11,10 +11,12 @@
 GODEPS += launchpad.net/go-dbus/v1
 GODEPS += launchpad.net/go-xdg/v0
 GODEPS += code.google.com/p/gosqlite/sqlite3
+GODEPS += launchpad.net/~ubuntu-push-hackers/ubuntu-push/go-uuid/uuid
 
 TOTEST = $(shell env GOPATH=$(GOPATH) go list $(PROJECT)/...|grep -v acceptance|grep -v http13client )
 
 bootstrap:
+	$(RM) -r $(GOPATH)/pkg
 	mkdir -p $(GOPATH)/bin
 	mkdir -p $(GOPATH)/pkg
 	go get -u launchpad.net/godeps
@@ -33,6 +35,7 @@
 
 build-client:
 	go build ubuntu-push-client.go
+	(cd signing-helper && cmake . && make)
 
 build-server-dev:
 	go build -o push-server-dev launchpad.net/ubuntu-push/server/dev

=== modified file 'README'
--- README	2014-03-31 17:58:54 +0000
+++ README	2014-05-12 14:11:46 +0000
@@ -6,11 +6,17 @@
 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:
+To setup Go dependencies, install the following dependencies:
+
+               build-essential
+               libsqlite3-dev
+
+and run:
 
    make bootstrap
 
-To run tests, install libgcrypt11-dev and libwhoopsie-dev and run:
+To run tests, install libglib2.0-dev, libgcrypt11-dev, libwhoopsie-dev,
+and run:
 
    make check
 

=== modified file 'bus/connectivity/connectivity.go'
--- bus/connectivity/connectivity.go	2014-04-04 11:08:28 +0000
+++ bus/connectivity/connectivity.go	2014-05-12 14:11:46 +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-05-12 14:11:46 +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,77 @@
 	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 }
+
+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-05-12 14:11:46 +0000
@@ -31,6 +31,7 @@
 
 // bus.Endpoint represents the DBus connection itself.
 type Endpoint interface {
+	GrabName(allowReplacement bool) <-chan error
 	WatchSignal(member string, f func(...interface{}), d func()) error
 	Call(member string, args []interface{}, rvs ...interface{}) error
 	GetProperty(property string) (interface{}, error)
@@ -53,7 +54,7 @@
 }
 
 // ensure endpoint implements Endpoint
-var _ Endpoint = &endpoint{}
+var _ Endpoint = (*endpoint)(nil)
 
 /*
    public methods
@@ -173,6 +174,19 @@
 	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.
+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
+}
+
 /*
    private methods
 */

=== 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-05-12 14:11:46 +0000
@@ -169,5 +169,16 @@
 // see Endpoint's Close. This one does nothing.
 func (tc *testingEndpoint) Close() {}
 
+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
+}
+
 // 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-05-12 14:11:46 +0000
@@ -173,3 +173,18 @@
 	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},
+		}})
+}

=== modified file 'client/client_test.go'
--- client/client_test.go	2014-04-18 16:31:04 +0000
+++ client/client_test.go	2014-05-12 14:11:46 +0000
@@ -265,8 +265,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{},
 	}
 	// sanity check that we are looking at all fields
 	vExpected := reflect.ValueOf(expected)
@@ -276,6 +277,8 @@
 		// field isn't empty/zero
 		c.Assert(fv.Interface(), Not(DeepEquals), reflect.Zero(fv.Type()).Interface(), Commentf("forgot about: %s", vExpected.Type().Field(i).Name))
 	}
+	// but AuthHelper really should be nil for now
+	expected.AuthHelper = nil
 	// finally compare
 	conf := cli.deriveSessionConfig(info)
 	c.Check(conf, DeepEquals, expected)

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-04-18 16:37:31 +0000
+++ client/session/session.go	2014-05-12 14:11:46 +0000
@@ -26,6 +26,7 @@
 	"fmt"
 	"math/rand"
 	"net"
+	"os/exec"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -38,7 +39,9 @@
 	"launchpad.net/ubuntu-push/util"
 )
 
-var wireVersionBytes = []byte{protocol.ProtocolWireVersion}
+var (
+	wireVersionBytes = []byte{protocol.ProtocolWireVersion}
+)
 
 type Notification struct {
 	TopLevel int64
@@ -84,6 +87,7 @@
 	ExpectAllRepairedTime  time.Duration
 	PEM                    []byte
 	Info                   map[string]interface{}
+	AuthHelper             []string
 }
 
 // ClientSession holds a client<->server session and its configuration.
@@ -115,6 +119,8 @@
 	stateP *uint32
 	ErrCh  chan error
 	MsgCh  chan *Notification
+	// authorization
+	auth string
 	// autoredial knobs
 	shouldDelayP    *uint32
 	lastAutoRedial  time.Time
@@ -234,6 +240,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
 }
@@ -457,10 +484,9 @@
 		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,13 +521,15 @@
 
 // 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 {
@@ -531,7 +559,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-05-12 14:11:46 +0000
@@ -34,7 +34,6 @@
 
 	"launchpad.net/ubuntu-push/client/gethosts"
 	"launchpad.net/ubuntu-push/client/session/levelmap"
-	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/protocol"
 	helpers "launchpad.net/ubuntu-push/testing"
 	"launchpad.net/ubuntu-push/testing/condition"
@@ -166,7 +165,7 @@
 /////
 
 type clientSessionSuite struct {
-	log  logger.Logger
+	log  *helpers.TestLogger
 	lvls func() (levelmap.LevelMap, error)
 }
 
@@ -347,6 +346,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
 ****************************************************************/
 
@@ -931,9 +967,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 +1075,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 +1092,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 +1115,7 @@
 	has_closed := false
 	err = sess.run(
 		func() { has_closed = true },
+		func() error { return nil },
 		func() error { return failure },
 		nil,
 		nil,
@@ -1076,6 +1131,7 @@
 	err = sess.run(
 		func() {},
 		func() error { return nil },
+		func() error { return nil },
 		func() error { return failure },
 		nil,
 		nil)
@@ -1090,6 +1146,7 @@
 		func() {},
 		func() error { return nil },
 		func() error { return nil },
+		func() error { return nil },
 		func() error { return failure },
 		nil)
 	c.Check(err, Equals, failure)
@@ -1109,6 +1166,7 @@
 		func() error { return nil },
 		func() error { return nil },
 		func() error { return nil },
+		func() error { return nil },
 		func() error { sess.MsgCh <- notf; return <-failureCh })
 	c.Check(err, Equals, nil)
 	// if run doesn't error it sets up the channels

=== modified file 'debian/changelog'
--- debian/changelog	2014-04-23 11:54:00 +0000
+++ debian/changelog	2014-05-12 14:11:46 +0000
@@ -1,3 +1,9 @@
+ubuntu-push (0.21-0.ubuntu1) UNRELEASED; urgency=medium
+
+  * New upstream release: first auth bits, and Qt dependency.
+
+ -- 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/control'
--- debian/control	2014-03-25 16:26:20 +0000
+++ debian/control	2014-05-12 14:11:46 +0000
@@ -14,6 +14,8 @@
                libgcrypt11-dev,
                libglib2.0-dev (>= 2.31.6),
                libwhoopsie-dev,
+               libubuntuoneauth-2.0-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-05-12 14:11:46 +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.install'
--- debian/ubuntu-push-client.install	2014-03-26 16:27:19 +0000
+++ debian/ubuntu-push-client.install	2014-05-12 14:11:46 +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-05-12 14:11:46 +0000
@@ -2,3 +2,4 @@
 launchpad.net/go-dbus/v1	bzr	[e-mail]-20140206110213-pbzcr6ucaz3rqmnw	125
 launchpad.net/go-xdg/v0	bzr	[e-mail]-20140208094800-gubd5md7cro3mtxa	10
 launchpad.net/gocheck	bzr	[e-mail]-20140127131816-zshobk1qqme626xw	86
+launchpad.net/~ubuntu-push-hackers/ubuntu-push/go-uuid	bzr	[e-mail]-20140130122455-pm9h8etl4owp90lg	1

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-04-04 13:54:45 +0000
+++ protocol/messages.go	2014-05-12 14:11:46 +0000
@@ -54,6 +54,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 +73,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"`
@@ -122,6 +154,17 @@
 	Notifications []Notification
 }
 
+// Reset resets the splitting state if the message storage is to be
+// reused.
+func (m *NotificationsMsg) Reset() {
+	// xxx
+}
+
+func (m *NotificationsMsg) Split() bool {
+	// xxx
+	return true
+}
+
 // A single unicast notification
 type Notification struct {
 	AppId string `json:"A"`
@@ -130,6 +173,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-05-12 14:11:46 +0000
@@ -104,6 +104,22 @@
 	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})
 }

=== 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-05-12 14:11:46 +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-05-12 14:11:46 +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-05-12 14:11:46 +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-05-12 14:11:46 +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>

=== 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-05-12 14:11:46 +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-05-12 14:11:46 +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":
+			events <- fmt.Sprintf("%swarn %s", sess.Prefix, recv.Reason)
 		}
 	}
 	return nil

=== 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-05-12 14:11:46 +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{
@@ -265,7 +262,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-05-12 14:11:46 +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
@@ -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-05-12 14:11:46 +0000
@@ -0,0 +1,96 @@
+/*
+ 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"
+	//"time"
+
+	. "launchpad.net/gocheck"
+
+	//"launchpad.net/ubuntu-push/protocol"
+	"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) {
+	userId, auth := s.associatedAuth("DEV1")
+	events, errCh, stop := s.StartClientAuth(c, "DEV1", nil, auth)
+	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.*")
+	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)
+}

=== modified file 'server/api/handlers.go'
--- server/api/handlers.go	2014-02-20 17:09:03 +0000
+++ server/api/handlers.go	2014-05-12 14:11:46 +0000
@@ -19,12 +19,15 @@
 package api
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"time"
 
+	"launchpad.net/~ubuntu-push-hackers/ubuntu-push/go-uuid/uuid"
+
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/store"
@@ -93,6 +96,11 @@
 		ioError,
 		"Could not read request body",
 	}
+	ErrMissingIdField = &APIError{
+		http.StatusBadRequest,
+		invalidRequest,
+		"Missing id field",
+	}
 	ErrMissingData = &APIError{
 		http.StatusBadRequest,
 		invalidRequest,
@@ -130,10 +138,17 @@
 	}
 )
 
-type Message struct {
-	Registration string          `json:"registration"`
-	CoalesceTag  string          `json:"coalesce_tag"`
-	Data         json.RawMessage `json:"data"`
+type castCommon struct {
+}
+
+type Unicast struct {
+	UserId   string `json:"userid"`
+	DeviceId string `json:"deviceid"`
+	AppId    string `json:"appid"`
+	//Registration string          `json:"registration"`
+	//CoalesceTag  string          `json:"coalesce_tag"`
+	ExpireOn string          `json:"expire_on"`
+	Data     json.RawMessage `json:"data"`
 }
 
 // Broadcast request JSON object.
@@ -198,11 +213,11 @@
 
 var zeroTime = time.Time{}
 
-func checkBroadcast(bcast *Broadcast) (time.Time, *APIError) {
-	if len(bcast.Data) == 0 {
+func checkCastCommon(data json.RawMessage, expireOn string) (time.Time, *APIError) {
+	if len(data) == 0 {
 		return zeroTime, ErrMissingData
 	}
-	expire, err := time.Parse(time.RFC3339, bcast.ExpireOn)
+	expire, err := time.Parse(time.RFC3339, expireOn)
 	if err != nil {
 		return zeroTime, ErrInvalidExpiration
 	}
@@ -212,6 +227,10 @@
 	return expire, nil
 }
 
+func checkBroadcast(bcast *Broadcast) (time.Time, *APIError) {
+	return checkCastCommon(bcast.Data, bcast.ExpireOn)
+}
+
 type StoreForRequest func(w http.ResponseWriter, request *http.Request) (store.PendingStore, error)
 
 // context holds the interfaces to delegate to serving requests
@@ -234,6 +253,20 @@
 	return sto, nil
 }
 
+func (ctx *context) prepare(w http.ResponseWriter, request *http.Request, reqObj interface{}) (store.PendingStore, *APIError) {
+	body, apiErr := ReadBody(request, MaxRequestBodyBytes)
+	if apiErr != nil {
+		return nil, apiErr
+	}
+
+	err := json.Unmarshal(body, reqObj)
+	if err != nil {
+		return nil, ErrMalformedJSONObject
+	}
+
+	return ctx.getStore(w, request)
+}
+
 type BroadcastHandler struct {
 	*context
 }
@@ -270,23 +303,13 @@
 		}
 	}()
 
-	body, apiErr := ReadBody(request, MaxRequestBodyBytes)
-	if apiErr != nil {
-		return
-	}
-
-	sto, apiErr := h.getStore(writer, request)
-	if apiErr != nil {
-		return
-	}
-	defer sto.Close()
-
 	broadcast := &Broadcast{}
-	err := json.Unmarshal(body, broadcast)
-	if err != nil {
-		apiErr = ErrMalformedJSONObject
+
+	sto, apiErr := h.prepare(writer, request, broadcast)
+	if apiErr != nil {
 		return
 	}
+	defer sto.Close()
 
 	apiErr = h.doBroadcast(sto, broadcast)
 	if apiErr != nil {
@@ -297,6 +320,64 @@
 	fmt.Fprintf(writer, `{"ok":true}`)
 }
 
+type UnicastHandler struct {
+	*context
+}
+
+func checkUnicast(ucast *Unicast) (time.Time, *APIError) {
+	if ucast.UserId == "" || ucast.DeviceId == "" || ucast.AppId == "" {
+		return zeroTime, ErrMissingIdField
+	}
+	return checkCastCommon(ucast.Data, ucast.ExpireOn)
+}
+
+// use a base64 encoded TimeUUID
+var generateMsgId = func() string {
+	return base64.StdEncoding.EncodeToString(uuid.NewUUID())
+}
+
+func (h *UnicastHandler) doUnicast(sto store.PendingStore, ucast *Unicast) *APIError {
+	expire, apiErr := checkUnicast(ucast)
+	if apiErr != nil {
+		return apiErr
+	}
+	chanId := store.UnicastInternalChannelId(ucast.UserId, ucast.DeviceId)
+	msgId := generateMsgId()
+	err := sto.AppendToUnicastChannel(chanId, ucast.AppId, ucast.Data, msgId, expire)
+	if err != nil {
+		h.logger.Errorf("could not store notification: %v", err)
+		return ErrCouldNotStoreNotification
+	}
+
+	h.broker.Unicast(chanId)
+	return nil
+}
+
+func (h *UnicastHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	var apiErr *APIError
+	defer func() {
+		if apiErr != nil {
+			RespondError(writer, apiErr)
+		}
+	}()
+
+	unicast := &Unicast{}
+
+	sto, apiErr := h.prepare(writer, request, unicast)
+	if apiErr != nil {
+		return
+	}
+	defer sto.Close()
+
+	apiErr = h.doUnicast(sto, unicast)
+	if apiErr != nil {
+		return
+	}
+
+	writer.Header().Set("Content-Type", "application/json")
+	fmt.Fprintf(writer, `{"ok":true}`)
+}
+
 // MakeHandlersMux makes a handler that dispatches for the various API endpoints.
 func MakeHandlersMux(storeForRequest StoreForRequest, broker broker.BrokerSending, logger logger.Logger) *http.ServeMux {
 	ctx := &context{
@@ -306,5 +387,6 @@
 	}
 	mux := http.NewServeMux()
 	mux.Handle("/broadcast", &BroadcastHandler{context: ctx})
+	mux.Handle("/notify", &UnicastHandler{context: ctx})
 	return mux
 }

=== modified file 'server/api/handlers_test.go'
--- server/api/handlers_test.go	2014-02-21 11:32:38 +0000
+++ server/api/handlers_test.go	2014-05-12 14:11:46 +0000
@@ -18,6 +18,7 @@
 
 import (
 	"bytes"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -30,8 +31,9 @@
 
 	. "launchpad.net/gocheck"
 
+	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/store"
-	helpers "launchpad.net/ubuntu-push/testing"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 func TestHandlers(t *testing.T) { TestingT(t) }
@@ -41,14 +43,14 @@
 	json            string
 	client          *http.Client
 	c               *C
-	testlog         *helpers.TestLogger
+	testlog         *help.TestLogger
 }
 
 var _ = Suite(&handlersSuite{})
 
 func (s *handlersSuite) SetUpTest(c *C) {
 	s.client = &http.Client{}
-	s.testlog = helpers.NewTestLogger(c, "error")
+	s.testlog = help.NewTestLogger(c, "error")
 }
 
 func (s *handlersSuite) TestAPIError(c *C) {
@@ -98,7 +100,7 @@
 
 var future = time.Now().Add(4 * time.Hour).Format(time.RFC3339)
 
-func (s *handlersSuite) TestCheckBroadcast(c *C) {
+func (s *handlersSuite) TestCheckCastBroadcastAndCommon(c *C) {
 	payload := json.RawMessage(`{"foo":"bar"}`)
 	broadcast := &Broadcast{
 		Channel:  "system",
@@ -134,19 +136,27 @@
 }
 
 type checkBrokerSending struct {
-	store    store.PendingStore
-	chanId   store.InternalChannelId
-	err      error
-	top      int64
-	payloads []json.RawMessage
+	store         store.PendingStore
+	chanId        store.InternalChannelId
+	err           error
+	top           int64
+	notifications []protocol.Notification
 }
 
 func (cbsend *checkBrokerSending) Broadcast(chanId store.InternalChannelId) {
-	top, payloads, err := cbsend.store.GetChannelSnapshot(chanId)
+	top, notifications, err := cbsend.store.GetChannelSnapshot(chanId)
 	cbsend.err = err
 	cbsend.chanId = chanId
 	cbsend.top = top
-	cbsend.payloads = payloads
+	cbsend.notifications = notifications
+}
+
+func (cbsend *checkBrokerSending) Unicast(chanIds ...store.InternalChannelId) {
+	// for now
+	if len(chanIds) != 1 {
+		panic("not expecting many chan ids for now")
+	}
+	cbsend.Broadcast(chanIds[0])
 }
 
 func (s *handlersSuite) TestDoBroadcast(c *C) {
@@ -163,7 +173,7 @@
 	c.Check(bsend.err, IsNil)
 	c.Check(bsend.chanId, Equals, store.SystemInternalChannelId)
 	c.Check(bsend.top, Equals, int64(1))
-	c.Check(bsend.payloads, DeepEquals, []json.RawMessage{payload})
+	c.Check(bsend.notifications, DeepEquals, help.Ns(payload))
 }
 
 func (s *handlersSuite) TestDoBroadcastUnknownChannel(c *C) {
@@ -192,6 +202,11 @@
 	return isto.intercept("AppendToChannel", err)
 }
 
+func (isto *interceptInMemoryPendingStore) AppendToUnicastChannel(chanId store.InternalChannelId, appId string, payload json.RawMessage, msgId string, expiration time.Time) error {
+	err := isto.InMemoryPendingStore.AppendToUnicastChannel(chanId, appId, payload, msgId, expiration)
+	return isto.intercept("AppendToUnicastChannel", err)
+}
+
 func (s *handlersSuite) TestDoBroadcastUnknownError(c *C) {
 	sto := &interceptInMemoryPendingStore{
 		store.NewInMemoryPendingStore(),
@@ -229,6 +244,115 @@
 	c.Check(s.testlog.Captured(), Equals, "ERROR could not store notification: fail\n")
 }
 
+func (s *handlersSuite) TestCheckUnicast(c *C) {
+	payload := json.RawMessage(`{"foo":"bar"}`)
+	unicast := func() *Unicast {
+		return &Unicast{
+			UserId:   "user1",
+			DeviceId: "DEV1",
+			AppId:    "app1",
+			ExpireOn: future,
+			Data:     payload,
+		}
+	}
+	u := unicast()
+	expire, apiErr := checkUnicast(u)
+	c.Assert(apiErr, IsNil)
+	c.Check(expire.Format(time.RFC3339), Equals, future)
+
+	u = unicast()
+	u.UserId = ""
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingIdField)
+
+	u = unicast()
+	u.AppId = ""
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingIdField)
+
+	u = unicast()
+	u.DeviceId = ""
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingIdField)
+
+	u = unicast()
+	u.Data = json.RawMessage(nil)
+	expire, apiErr = checkUnicast(u)
+	c.Check(apiErr, Equals, ErrMissingData)
+}
+
+func (s *handlersSuite) TestGenerateMsgId(c *C) {
+	msgId := generateMsgId()
+	decoded, err := base64.StdEncoding.DecodeString(msgId)
+	c.Assert(err, IsNil)
+	c.Check(decoded, HasLen, 16)
+}
+
+func (s *handlersSuite) TestDoUnicast(c *C) {
+	prevGenMsgId := generateMsgId
+	defer func() {
+		generateMsgId = prevGenMsgId
+	}()
+	generateMsgId = func() string {
+		return "MSG-ID"
+	}
+	sto := store.NewInMemoryPendingStore()
+	bsend := &checkBrokerSending{store: sto}
+	bh := &UnicastHandler{&context{nil, bsend, nil}}
+	payload := json.RawMessage(`{"a": 1}`)
+	apiErr := bh.doUnicast(sto, &Unicast{
+		UserId:   "user1",
+		DeviceId: "DEV1",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     payload,
+	})
+	c.Check(apiErr, IsNil)
+	c.Check(bsend.err, IsNil)
+	c.Check(bsend.chanId, Equals, store.UnicastInternalChannelId("user1", "DEV1"))
+	c.Check(bsend.top, Equals, int64(0))
+	c.Check(bsend.notifications, DeepEquals, []protocol.Notification{
+		protocol.Notification{
+			AppId:   "app1",
+			MsgId:   "MSG-ID",
+			Payload: payload,
+		},
+	})
+}
+
+func (s *handlersSuite) TestDoUnicastMissingIdField(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	bh := &UnicastHandler{}
+	apiErr := bh.doUnicast(sto, &Unicast{
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrMissingIdField)
+}
+
+func (s *handlersSuite) TestDoUnicastCouldNotStoreNotification(c *C) {
+	sto := &interceptInMemoryPendingStore{
+		store.NewInMemoryPendingStore(),
+		func(meth string, err error) error {
+			if meth == "AppendToUnicastChannel" {
+				return errors.New("fail")
+			}
+			return err
+		},
+	}
+	ctx := &context{logger: s.testlog}
+	bh := &UnicastHandler{ctx}
+	apiErr := bh.doUnicast(sto, &Unicast{
+		UserId:   "user1",
+		DeviceId: "DEV1",
+		AppId:    "app1",
+		ExpireOn: future,
+		Data:     json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrCouldNotStoreNotification)
+	c.Check(s.testlog.Captured(), Equals, "ERROR could not store notification: fail\n")
+}
+
 func newPostRequest(path string, message interface{}, server *httptest.Server) *http.Request {
 	packedMessage, err := json.Marshal(message)
 	if err != nil {
@@ -268,6 +392,14 @@
 	bsend.chanId <- chanId
 }
 
+func (bsend testBrokerSending) Unicast(chanIds ...store.InternalChannelId) {
+	// for now
+	if len(chanIds) != 1 {
+		panic("not expecting many chan ids for now")
+	}
+	bsend.chanId <- chanIds[0]
+}
+
 func (s *handlersSuite) TestRespondsToBasicSystemBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
@@ -480,3 +612,73 @@
 
 	checkError(c, response, ErrWrongRequestMethod)
 }
+
+func (s *handlersSuite) TestRespondsUnicast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return sto, nil
+	}
+	bsend := testBrokerSending{make(chan store.InternalChannelId, 1)}
+	testServer := httptest.NewServer(MakeHandlersMux(stoForReq, bsend, nil))
+	defer testServer.Close()
+
+	payload := json.RawMessage(`{"foo":"bar"}`)
+
+	request := newPostRequest("/notify", &Unicast{
+		UserId:   "user2",
+		DeviceId: "dev3",
+		AppId:    "app2",
+		ExpireOn: future,
+		Data:     payload,
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+
+	c.Check(response.StatusCode, Equals, http.StatusOK)
+	c.Check(response.Header.Get("Content-Type"), Equals, "application/json")
+	body, err := getResponseBody(response)
+	c.Assert(err, IsNil)
+	c.Check(string(body), Matches, ".*ok.*")
+
+	chanId := store.UnicastInternalChannelId("user2", "dev3")
+	c.Check(<-bsend.chanId, Equals, chanId)
+	top, notifications, err := sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+	c.Check(top, Equals, int64(0))
+	c.Check(notifications, HasLen, 1)
+}
+
+func (s *handlersSuite) TestCannotUnicastTooBigMessages(c *C) {
+	testServer := httptest.NewServer(&UnicastHandler{})
+	defer testServer.Close()
+
+	bigString := strings.Repeat("a", MaxRequestBodyBytes)
+	dataString := fmt.Sprintf(`"%v"`, bigString)
+
+	request := newPostRequest("/", &Unicast{
+		ExpireOn: future,
+		Data:     json.RawMessage([]byte(dataString)),
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrRequestBodyTooLarge)
+}
+
+func (s *handlersSuite) TestCannotUnicastWithMissingFields(c *C) {
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return store.NewInMemoryPendingStore(), nil
+	}
+	ctx := &context{stoForReq, nil, nil}
+	testServer := httptest.NewServer(&UnicastHandler{ctx})
+	defer testServer.Close()
+
+	request := newPostRequest("/", &Unicast{
+		Data: json.RawMessage(`{"foo":"bar"}`),
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrMissingIdField)
+}

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-04-04 09:58:34 +0000
+++ server/broker/broker.go	2014-05-12 14:11:46 +0000
@@ -30,7 +30,7 @@
 // through them.
 type Broker interface {
 	// Register the session.
-	Register(*protocol.ConnectMsg) (BrokerSession, error)
+	Register(connMsg *protocol.ConnectMsg, sessionId string) (BrokerSession, error)
 	// Unregister the session.
 	Unregister(BrokerSession)
 }
@@ -39,6 +39,8 @@
 type BrokerSending interface {
 	// Broadcast channel.
 	Broadcast(chanId store.InternalChannelId)
+	// Unicast over channels.
+	Unicast(chanIds ...store.InternalChannelId)
 }
 
 // Exchange leads the session through performing an exchange, typically delivery.
@@ -81,6 +83,10 @@
 	Levels() LevelsMap
 	// ExchangeScratchArea returns the scratch area for exchanges.
 	ExchangeScratchArea() *ExchangesScratchArea
+	// Get gets the content of the channel with chanId.
+	Get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error)
+	// DropByMsgId drops notifications from the channel chanId by message id.
+	DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error
 }
 
 // Session aborted error.

=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-04-04 13:19:10 +0000
+++ server/broker/exchanges.go	2014-05-12 14:11:46 +0000
@@ -28,18 +28,18 @@
 
 // Scratch area for exchanges, sessions should hold one of these.
 type ExchangesScratchArea struct {
-	broadcastMsg  protocol.BroadcastMsg
-	ackMsg        protocol.AckMsg
-	connBrokenMsg protocol.ConnBrokenMsg
+	broadcastMsg     protocol.BroadcastMsg
+	notificationsMsg protocol.NotificationsMsg
+	ackMsg           protocol.AckMsg
 }
 
 // BroadcastExchange leads a session through delivering a BROADCAST.
 // For simplicity it is fully public.
 type BroadcastExchange struct {
-	ChanId               store.InternalChannelId
-	TopLevel             int64
-	NotificationPayloads []json.RawMessage
-	Decoded              []map[string]interface{}
+	ChanId        store.InternalChannelId
+	TopLevel      int64
+	Notifications []protocol.Notification
+	Decoded       []map[string]interface{}
 }
 
 // check interface already here
@@ -47,18 +47,18 @@
 
 // Init ensures the BroadcastExchange is fully initialized for the sessions.
 func (sbe *BroadcastExchange) Init() {
-	decoded := make([]map[string]interface{}, len(sbe.NotificationPayloads))
+	decoded := make([]map[string]interface{}, len(sbe.Notifications))
 	sbe.Decoded = decoded
-	for i, p := range sbe.NotificationPayloads {
-		err := json.Unmarshal(p, &decoded[i])
+	for i, notif := range sbe.Notifications {
+		err := json.Unmarshal(notif.Payload, &decoded[i])
 		if err != nil {
 			decoded[i] = nil
 		}
 	}
 }
 
-func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
-	c := int64(len(payloads))
+func filterByLevel(clientLevel, topLevel int64, notifs []protocol.Notification) []protocol.Notification {
+	c := int64(len(notifs))
 	if c == 0 {
 		return nil
 	}
@@ -67,32 +67,32 @@
 		delta = 1
 	}
 	if delta < c {
-		return payloads[c-delta:]
+		return notifs[c-delta:]
 	} else {
-		return payloads
+		return notifs
 	}
 }
 
-func channelFilter(tag string, chanId store.InternalChannelId, payloads []json.RawMessage, decoded []map[string]interface{}) []json.RawMessage {
-	if len(payloads) != 0 && chanId == store.SystemInternalChannelId {
-		decoded := decoded[len(decoded)-len(payloads):]
+func channelFilter(tag string, chanId store.InternalChannelId, notifs []protocol.Notification, decoded []map[string]interface{}) []json.RawMessage {
+	if len(notifs) != 0 && chanId == store.SystemInternalChannelId {
+		decoded := decoded[len(decoded)-len(notifs):]
 		filtered := make([]json.RawMessage, 0)
 		for i, decoded1 := range decoded {
 			if _, ok := decoded1[tag]; ok {
-				filtered = append(filtered, payloads[i])
+				filtered = append(filtered, notifs[i].Payload)
 			}
 		}
-		payloads = filtered
+		return filtered
 	}
-	return payloads
+	return protocol.ExtractPayloads(notifs)
 }
 
 // Prepare session for a BROADCAST.
 func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
 	clientLevel := sess.Levels()[sbe.ChanId]
-	payloads := filterByLevel(clientLevel, sbe.TopLevel, sbe.NotificationPayloads)
+	notifs := filterByLevel(clientLevel, sbe.TopLevel, sbe.Notifications)
 	tag := fmt.Sprintf("%s/%s", sess.DeviceImageChannel(), sess.DeviceImageModel())
-	payloads = channelFilter(tag, sbe.ChanId, payloads, sbe.Decoded)
+	payloads := channelFilter(tag, sbe.ChanId, notifs, sbe.Decoded)
 	if len(payloads) == 0 && sbe.TopLevel >= clientLevel {
 		// empty and don't need to force resync => do nothing
 		return nil, nil, ErrNop
@@ -119,23 +119,55 @@
 	return nil
 }
 
-// ConnBrokenExchange breaks a session giving a reason.
-type ConnBrokenExchange struct {
-	Reason string
-}
-
-// check interface already here
-var _ Exchange = (*ConnBrokenExchange)(nil)
-
-// Prepare session for a CONNBROKEN.
-func (cbe *ConnBrokenExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
-	scratchArea := sess.ExchangeScratchArea()
-	scratchArea.connBrokenMsg.Type = "connbroken"
-	scratchArea.connBrokenMsg.Reason = cbe.Reason
-	return &scratchArea.connBrokenMsg, nil, nil
-}
-
-// CONNBROKEN isn't acked
-func (cbe *ConnBrokenExchange) Acked(sess BrokerSession, done bool) error {
-	panic("Acked should not get invoked on ConnBrokenExchange")
+// ConnMetaExchange allows to send a CONNBROKEN or CONNWARN message.
+type ConnMetaExchange struct {
+	Msg protocol.OnewayMsg
+}
+
+// check interface already here
+var _ Exchange = (*ConnMetaExchange)(nil)
+
+// Prepare session for a CONNBROKEN/WARN.
+func (cbe *ConnMetaExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
+	return cbe.Msg, nil, nil
+}
+
+// CONNBROKEN/WARN aren't acked.
+func (cbe *ConnMetaExchange) Acked(sess BrokerSession, done bool) error {
+	panic("Acked should not get invoked on ConnMetaExchange")
+}
+
+// UnicastExchange leads a session through delivering a NOTIFICATIONS message.
+// For simplicity it is fully public.
+type UnicastExchange struct {
+	ChanId store.InternalChannelId
+}
+
+// check interface already here
+var _ Exchange = (*UnicastExchange)(nil)
+
+// Prepare session for a NOTIFICATIONS.
+func (sue *UnicastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
+	_, notifs, err := sess.Get(sue.ChanId, false)
+	if err != nil {
+		return nil, nil, err
+	}
+	scratchArea := sess.ExchangeScratchArea()
+	scratchArea.notificationsMsg.Reset()
+	scratchArea.notificationsMsg.Type = "notifications"
+	scratchArea.notificationsMsg.Notifications = notifs
+	return &scratchArea.notificationsMsg, &scratchArea.ackMsg, nil
+}
+
+// Acked deals with an ACK for a NOTIFICATIONS.
+func (sue *UnicastExchange) Acked(sess BrokerSession, done bool) error {
+	scratchArea := sess.ExchangeScratchArea()
+	if scratchArea.ackMsg.Type != "ack" {
+		return &ErrAbort{"expected ACK message"}
+	}
+	err := sess.DropByMsgId(sue.ChanId, scratchArea.notificationsMsg.Notifications)
+	if err != nil {
+		return err
+	}
+	return nil
 }

=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-04-04 13:19:10 +0000
+++ server/broker/exchanges_test.go	2014-05-12 14:11:46 +0000
@@ -18,15 +18,18 @@
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	stdtesting "testing"
 
 	. "launchpad.net/gocheck"
 
+	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/broker/testing"
 	"launchpad.net/ubuntu-push/server/store"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 func TestBroker(t *stdtesting.T) { TestingT(t) }
@@ -39,11 +42,11 @@
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
-		NotificationPayloads: []json.RawMessage{
+		Notifications: help.Ns(
 			json.RawMessage(`{"a":"x"}`),
 			json.RawMessage(`[]`),
 			json.RawMessage(`{"a":"y"}`),
-		},
+		),
 	}
 	exchg.Init()
 	c.Check(exchg.Decoded, DeepEquals, []map[string]interface{}{
@@ -62,10 +65,10 @@
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
-		NotificationPayloads: []json.RawMessage{
+		Notifications: help.Ns(
 			json.RawMessage(`{"img1/m1":100}`),
 			json.RawMessage(`{"img2/m2":200}`),
-		},
+		),
 	}
 	exchg.Init()
 	outMsg, inMsg, err := exchg.Prepare(sess)
@@ -88,9 +91,9 @@
 		ImageChannel: "img1",
 	}
 	exchg := &broker.BroadcastExchange{
-		ChanId:               store.SystemInternalChannelId,
-		TopLevel:             3,
-		NotificationPayloads: []json.RawMessage{},
+		ChanId:        store.SystemInternalChannelId,
+		TopLevel:      3,
+		Notifications: []protocol.Notification{},
 	}
 	exchg.Init()
 	outMsg, inMsg, err := exchg.Prepare(sess)
@@ -108,9 +111,9 @@
 		ImageChannel: "img1",
 	}
 	exchg := &broker.BroadcastExchange{
-		ChanId:               store.SystemInternalChannelId,
-		TopLevel:             3,
-		NotificationPayloads: []json.RawMessage{},
+		ChanId:        store.SystemInternalChannelId,
+		TopLevel:      3,
+		Notifications: []protocol.Notification{},
 	}
 	exchg.Init()
 	outMsg, inMsg, err := exchg.Prepare(sess)
@@ -133,9 +136,9 @@
 
 	topLevel := int64(len(needsSplitting))
 	exchg := &broker.BroadcastExchange{
-		ChanId:               store.SystemInternalChannelId,
-		TopLevel:             topLevel,
-		NotificationPayloads: needsSplitting,
+		ChanId:        store.SystemInternalChannelId,
+		TopLevel:      topLevel,
+		Notifications: help.Ns(needsSplitting...),
 	}
 	exchg.Init()
 	outMsg, _, err := exchg.Prepare(sess)
@@ -152,10 +155,10 @@
 	exchg = &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: topLevel + 2,
-		NotificationPayloads: []json.RawMessage{
+		Notifications: help.Ns(
 			json.RawMessage(`{"img1/m1":"x"}`),
 			json.RawMessage(`{"img1/m1":"y"}`),
-		},
+		),
 	}
 	exchg.Init()
 	outMsg, _, err = exchg.Prepare(sess)
@@ -173,9 +176,9 @@
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
-		NotificationPayloads: []json.RawMessage{
+		Notifications: help.Ns(
 			json.RawMessage(`{"img2/m1":1}`),
-		},
+		),
 	}
 	exchg.Init()
 	outMsg, inMsg, err := exchg.Prepare(sess)
@@ -202,10 +205,10 @@
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 3,
-		NotificationPayloads: []json.RawMessage{
+		Notifications: help.Ns(
 			json.RawMessage(`{"img1/m1":100}`),
 			json.RawMessage(`{"img1/m1":101}`),
-		},
+		),
 	}
 	exchg.Init()
 	outMsg, inMsg, err := exchg.Prepare(sess)
@@ -229,11 +232,11 @@
 	exchg := &broker.BroadcastExchange{
 		ChanId:   store.SystemInternalChannelId,
 		TopLevel: 5,
-		NotificationPayloads: []json.RawMessage{
+		Notifications: help.Ns(
 			json.RawMessage(`{"img1/m1":100}`),
 			json.RawMessage(`{"img2/m2":200}`),
 			json.RawMessage(`{"img1/m1":101}`),
-		},
+		),
 	}
 	exchg.Init()
 	outMsg, inMsg, err := exchg.Prepare(sess)
@@ -249,16 +252,114 @@
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(5))
 }
 
-func (s *exchangesSuite) TestConnBrokenExchange(c *C) {
+func (s *exchangesSuite) TestConnMetaExchange(c *C) {
 	sess := &testing.TestBrokerSession{}
-	cbe := &broker.ConnBrokenExchange{"REASON"}
+	var msg protocol.OnewayMsg = &protocol.ConnWarnMsg{"connwarn", "REASON"}
+	cbe := &broker.ConnMetaExchange{msg}
 	outMsg, inMsg, err := cbe.Prepare(sess)
 	c.Assert(err, IsNil)
+	c.Check(msg, Equals, outMsg)
 	c.Check(inMsg, IsNil) // no answer is expected
 	// check
 	marshalled, err := json.Marshal(outMsg)
 	c.Assert(err, IsNil)
-	c.Check(string(marshalled), Equals, `{"T":"connbroken","Reason":"REASON"}`)
-
-	c.Check(func() { cbe.Acked(nil, true) }, PanicMatches, "Acked should not get invoked on ConnBrokenExchange")
+	c.Check(string(marshalled), Equals, `{"T":"connwarn","Reason":"REASON"}`)
+
+	c.Check(func() { cbe.Acked(nil, true) }, PanicMatches, "Acked should not get invoked on ConnMetaExchange")
+}
+
+func (s *exchangesSuite) TestUnicastExchange(c *C) {
+	chanId1 := store.UnicastInternalChannelId("u1", "d1")
+	notifs := []protocol.Notification{
+		protocol.Notification{
+			MsgId:   "msg1",
+			AppId:   "app1",
+			Payload: json.RawMessage(`{"m": 1}`),
+		},
+		protocol.Notification{
+			MsgId:   "msg2",
+			AppId:   "app2",
+			Payload: json.RawMessage(`{"m": 2}`),
+		},
+	}
+	dropped := make(chan []protocol.Notification, 2)
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			c.Check(chanId, Equals, chanId1)
+			c.Check(cachedOk, Equals, false)
+			return 0, notifs, nil
+		},
+		DoDropByMsgId: func(chanId store.InternalChannelId, targets []protocol.Notification) error {
+			c.Check(chanId, Equals, chanId1)
+			dropped <- targets
+			return nil
+		},
+	}
+	exchg := &broker.UnicastExchange{chanId1}
+	outMsg, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(outMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"notifications","Notifications":[{"A":"app1","M":"msg1","P":{"m":1}},{"A":"app2","M":"msg2","P":{"m":2}}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, IsNil)
+	c.Check(dropped, HasLen, 1)
+	c.Check(<-dropped, DeepEquals, notifs)
+}
+
+func (s *exchangesSuite) TestUnicastExchangeAckMismatch(c *C) {
+	notifs := []protocol.Notification{}
+	dropped := make(chan []protocol.Notification, 2)
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			return 0, notifs, nil
+		},
+		DoDropByMsgId: func(chanId store.InternalChannelId, targets []protocol.Notification) error {
+			dropped <- targets
+			return nil
+		},
+	}
+	exchg := &broker.UnicastExchange{}
+	_, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	err = json.Unmarshal([]byte(`{}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, Not(IsNil))
+	c.Check(dropped, HasLen, 0)
+}
+
+func (s *exchangesSuite) TestUnicastExchangeErrorOnPrepare(c *C) {
+	fail := errors.New("fail")
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			return 0, nil, fail
+		},
+	}
+	exchg := &broker.UnicastExchange{}
+	_, _, err := exchg.Prepare(sess)
+	c.Assert(err, Equals, fail)
+}
+
+func (s *exchangesSuite) TestUnicastExchangeErrorOnAcked(c *C) {
+	notifs := []protocol.Notification{}
+	fail := errors.New("fail")
+	sess := &testing.TestBrokerSession{
+		DoGet: func(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+			return 0, notifs, nil
+		},
+		DoDropByMsgId: func(chanId store.InternalChannelId, targets []protocol.Notification) error {
+			return fail
+		},
+	}
+	exchg := &broker.UnicastExchange{}
+	_, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess, true)
+	c.Assert(err, Equals, fail)
 }

=== modified file 'server/broker/exchg_impl_test.go'
--- server/broker/exchg_impl_test.go	2014-04-03 16:00:53 +0000
+++ server/broker/exchg_impl_test.go	2014-05-12 14:11:46 +0000
@@ -22,6 +22,7 @@
 	. "launchpad.net/gocheck"
 
 	"launchpad.net/ubuntu-push/server/store"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 type exchangesImplSuite struct{}
@@ -29,27 +30,27 @@
 var _ = Suite(&exchangesImplSuite{})
 
 func (s *exchangesImplSuite) TestFilterByLevel(c *C) {
-	payloads := []json.RawMessage{
+	notifs := help.Ns(
 		json.RawMessage(`{"a": 3}`),
 		json.RawMessage(`{"a": 4}`),
 		json.RawMessage(`{"a": 5}`),
-	}
-	res := filterByLevel(5, 5, payloads)
+	)
+	res := filterByLevel(5, 5, notifs)
 	c.Check(len(res), Equals, 0)
-	res = filterByLevel(4, 5, payloads)
+	res = filterByLevel(4, 5, notifs)
 	c.Check(len(res), Equals, 1)
-	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 5}`))
-	res = filterByLevel(3, 5, payloads)
+	c.Check(res[0].Payload, DeepEquals, json.RawMessage(`{"a": 5}`))
+	res = filterByLevel(3, 5, notifs)
 	c.Check(len(res), Equals, 2)
-	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 4}`))
-	res = filterByLevel(2, 5, payloads)
+	c.Check(res[0].Payload, DeepEquals, json.RawMessage(`{"a": 4}`))
+	res = filterByLevel(2, 5, notifs)
 	c.Check(len(res), Equals, 3)
-	res = filterByLevel(1, 5, payloads)
+	res = filterByLevel(1, 5, notifs)
 	c.Check(len(res), Equals, 3)
 	// too ahead, pick only last
-	res = filterByLevel(10, 5, payloads)
+	res = filterByLevel(10, 5, notifs)
 	c.Check(len(res), Equals, 1)
-	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 5}`))
+	c.Check(res[0].Payload, DeepEquals, json.RawMessage(`{"a": 5}`))
 }
 
 func (s *exchangesImplSuite) TestFilterByLevelEmpty(c *C) {
@@ -71,18 +72,19 @@
 		err := json.Unmarshal(p, &decoded[i])
 		c.Assert(err, IsNil)
 	}
+	notifs := help.Ns(payloads...)
 
 	other := store.InternalChannelId("1")
 
 	c.Check(channelFilter("", store.SystemInternalChannelId, nil, nil), IsNil)
-	c.Check(channelFilter("", other, payloads[1:], decoded), DeepEquals, payloads[1:])
+	c.Check(channelFilter("", other, notifs[1:], decoded), DeepEquals, payloads[1:])
 
 	// use tag when channel is the sytem channel
 
-	c.Check(channelFilter("c/z", store.SystemInternalChannelId, payloads, decoded), HasLen, 0)
-
-	c.Check(channelFilter("a/x", store.SystemInternalChannelId, payloads, decoded), DeepEquals, []json.RawMessage{payloads[0], payloads[3]})
-
-	c.Check(channelFilter("a/x", store.SystemInternalChannelId, payloads[1:], decoded), DeepEquals, []json.RawMessage{payloads[3]})
+	c.Check(channelFilter("c/z", store.SystemInternalChannelId, notifs, decoded), HasLen, 0)
+
+	c.Check(channelFilter("a/x", store.SystemInternalChannelId, notifs, decoded), DeepEquals, []json.RawMessage{payloads[0], payloads[3]})
+
+	c.Check(channelFilter("a/x", store.SystemInternalChannelId, notifs[1:], decoded), DeepEquals, []json.RawMessage{payloads[3]})
 
 }

=== modified file 'server/broker/simple/simple.go'
--- server/broker/simple/simple.go	2014-04-11 08:47:18 +0000
+++ server/broker/simple/simple.go	2014-05-12 14:11:46 +0000
@@ -46,6 +46,7 @@
 
 // simpleBrokerSession represents a session in the broker.
 type simpleBrokerSession struct {
+	broker       *SimpleBroker
 	registered   bool
 	deviceId     string
 	model        string
@@ -61,6 +62,7 @@
 
 const (
 	broadcastDelivery deliveryKind = iota
+	unicastDelivery
 )
 
 // delivery holds all the information to request a delivery
@@ -93,6 +95,14 @@
 	return &sess.exchgScratch
 }
 
+func (sess *simpleBrokerSession) Get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+	return sess.broker.get(chanId, cachedOk)
+}
+
+func (sess *simpleBrokerSession) DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	return sess.broker.drop(chanId, targets)
+}
+
 // NewSimpleBroker makes a new SimpleBroker.
 func NewSimpleBroker(sto store.PendingStore, cfg broker.BrokerConfig, logger logger.Logger) *SimpleBroker {
 	sessionCh := make(chan *simpleBrokerSession, cfg.BrokerQueueSize())
@@ -144,7 +154,7 @@
 	// find relevant channels, for now only system
 	channels := []store.InternalChannelId{store.SystemInternalChannelId}
 	for _, chanId := range channels {
-		topLevel, payloads, err := b.sto.GetChannelSnapshot(chanId)
+		topLevel, notifications, err := b.sto.GetChannelSnapshot(chanId)
 		if err != nil {
 			// next broadcast will try again
 			b.logger.Errorf("unsuccessful feed pending, get channel snapshot for %v: %v", chanId, err)
@@ -153,9 +163,9 @@
 		clientLevel := sess.levels[chanId]
 		if clientLevel != topLevel {
 			broadcastExchg := &broker.BroadcastExchange{
-				ChanId:               chanId,
-				TopLevel:             topLevel,
-				NotificationPayloads: payloads,
+				ChanId:        chanId,
+				TopLevel:      topLevel,
+				Notifications: notifications,
 			}
 			broadcastExchg.Init()
 			sess.exchanges <- broadcastExchg
@@ -166,7 +176,7 @@
 
 // Register registers a session with the broker. It feeds the session
 // pending notifications as well.
-func (b *SimpleBroker) Register(connect *protocol.ConnectMsg) (broker.BrokerSession, error) {
+func (b *SimpleBroker) Register(connect *protocol.ConnectMsg, sessionId string) (broker.BrokerSession, error) {
 	// xxx sanity check DeviceId
 	model, err := broker.GetInfoString(connect, "device", "?")
 	if err != nil {
@@ -185,6 +195,7 @@
 		levels[id] = v
 	}
 	sess := &simpleBrokerSession{
+		broker:       b,
 		deviceId:     connect.DeviceId,
 		model:        model,
 		imageChannel: imageChannel,
@@ -207,6 +218,24 @@
 	b.sessionCh <- sess
 }
 
+func (b *SimpleBroker) get(chanId store.InternalChannelId, cachedOk bool) (int64, []protocol.Notification, error) {
+	topLevel, notifications, err := b.sto.GetChannelSnapshot(chanId)
+	if err != nil {
+		b.logger.Errorf("unsuccessful, get channel snapshot for %v (cachedOk=%v): %v", chanId, cachedOk, err)
+	}
+	return topLevel, notifications, err
+
+}
+
+func (b *SimpleBroker) drop(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	err := b.sto.DropByMsgId(chanId, targets)
+	if err != nil {
+		b.logger.Errorf("unsuccessful, drop from channel %v: %v", chanId, err)
+	}
+	return err
+
+}
+
 // run runs the agent logic of the broker.
 func (b *SimpleBroker) run() {
 Loop:
@@ -224,7 +253,7 @@
 			} else { // register
 				prev := b.registry[sess.deviceId]
 				if prev != nil { // kick it
-					close(prev.exchanges)
+					prev.exchanges <- nil
 				}
 				b.registry[sess.deviceId] = sess
 				sess.registered = true
@@ -233,21 +262,27 @@
 		case delivery := <-b.deliveryCh:
 			switch delivery.kind {
 			case broadcastDelivery:
-				topLevel, payloads, err := b.sto.GetChannelSnapshot(delivery.chanId)
+				topLevel, notifications, err := b.get(delivery.chanId, false)
 				if err != nil {
 					// next broadcast will try again
-					b.logger.Errorf("unsuccessful broadcast, get channel snapshot for %v: %v", delivery.chanId, err)
 					continue Loop
 				}
 				broadcastExchg := &broker.BroadcastExchange{
-					ChanId:               delivery.chanId,
-					TopLevel:             topLevel,
-					NotificationPayloads: payloads,
+					ChanId:        delivery.chanId,
+					TopLevel:      topLevel,
+					Notifications: notifications,
 				}
 				broadcastExchg.Init()
 				for _, sess := range b.registry {
 					sess.exchanges <- broadcastExchg
 				}
+			case unicastDelivery:
+				chanId := delivery.chanId
+				_, devId := chanId.UnicastUserAndDevice()
+				sess := b.registry[devId]
+				if sess != nil {
+					sess.exchanges <- &broker.UnicastExchange{chanId}
+				}
 			}
 		}
 	}
@@ -260,3 +295,13 @@
 		chanId: chanId,
 	}
 }
+
+// Unicast requests unicast for the channels.
+func (b *SimpleBroker) Unicast(chanIds ...store.InternalChannelId) {
+	for _, chanId := range chanIds {
+		b.deliveryCh <- &delivery{
+			kind:   unicastDelivery,
+			chanId: chanId,
+		}
+	}
+}

=== modified file 'server/broker/simple/simple_test.go'
--- server/broker/simple/simple_test.go	2014-04-03 16:00:53 +0000
+++ server/broker/simple/simple_test.go	2014-05-12 14:11:46 +0000
@@ -26,6 +26,7 @@
 	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/broker/testing"
 	"launchpad.net/ubuntu-push/server/store"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 func TestSimple(t *stdtesting.T) { TestingT(t) }
@@ -58,10 +59,10 @@
 	c.Assert(len(sess.exchanges), Equals, 1)
 	exchg1 := <-sess.exchanges
 	c.Check(exchg1, DeepEquals, &broker.BroadcastExchange{
-		ChanId:               store.SystemInternalChannelId,
-		TopLevel:             1,
-		NotificationPayloads: []json.RawMessage{notification1},
-		Decoded:              []map[string]interface{}{decoded1},
+		ChanId:        store.SystemInternalChannelId,
+		TopLevel:      1,
+		Notifications: help.Ns(notification1),
+		Decoded:       []map[string]interface{}{decoded1},
 	})
 }
 

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

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

=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-04-11 08:47:18 +0000
+++ server/broker/testsuite/suite.go	2014-05-12 14:11:46 +0000
@@ -30,7 +30,7 @@
 	"launchpad.net/ubuntu-push/server/broker"
 	"launchpad.net/ubuntu-push/server/broker/testing"
 	"launchpad.net/ubuntu-push/server/store"
-	helpers "launchpad.net/ubuntu-push/testing"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 // The expected interface for tested brokers.
@@ -50,12 +50,14 @@
 	RevealSession func(broker.Broker, string) broker.BrokerSession
 	// Let us get to a broker.BroadcastExchange from an Exchange.
 	RevealBroadcastExchange func(broker.Exchange) *broker.BroadcastExchange
+	// Let us get to a broker.UnicastExchange from an Exchange.
+	RevealUnicastExchange func(broker.Exchange) *broker.UnicastExchange
 	// private
-	testlog *helpers.TestLogger
+	testlog *help.TestLogger
 }
 
 func (s *CommonBrokerSuite) SetUpTest(c *C) {
-	s.testlog = helpers.NewTestLogger(c, "error")
+	s.testlog = help.NewTestLogger(c, "error")
 }
 
 var testBrokerConfig = &testing.TestBrokerConfig{10, 5}
@@ -89,7 +91,7 @@
 			"device":  "model",
 			"channel": "daily",
 		},
-	})
+	}, "s1")
 	c.Assert(err, IsNil)
 	c.Assert(s.RevealSession(b, "dev-1"), Equals, sess)
 	c.Assert(sess.DeviceIdentifier(), Equals, "dev-1")
@@ -101,7 +103,7 @@
 	}))
 	b.Unregister(sess)
 	// just to make sure the unregister was processed
-	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""})
+	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""}, "s2")
 	c.Assert(err, IsNil)
 	c.Check(s.RevealSession(b, "dev-1"), IsNil)
 }
@@ -111,7 +113,7 @@
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
-	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1", Levels: map[string]int64{"z": 5}})
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1", Levels: map[string]int64{"z": 5}}, "s1")
 	c.Check(err, FitsTypeOf, &broker.ErrAbort{})
 }
 
@@ -123,11 +125,11 @@
 	info := map[string]interface{}{
 		"device": -1,
 	}
-	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", Info: info})
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", Info: info}, "s1")
 	c.Check(err, Equals, broker.ErrUnexpectedValue)
 	info["device"] = "m"
 	info["channel"] = -1
-	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", Info: info})
+	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", Info: info}, "s2")
 	c.Check(err, Equals, broker.ErrUnexpectedValue)
 }
 
@@ -139,7 +141,7 @@
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
-	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
 	c.Check(len(sess.SessionChannel()), Equals, 1)
 }
@@ -149,7 +151,7 @@
 	b := s.MakeBroker(sto, testBrokerConfig, s.testlog)
 	b.Start()
 	defer b.Stop()
-	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
 	// but
 	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful feed pending, get channel snapshot for 0: get channel snapshot fail\n")
@@ -160,22 +162,25 @@
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
-	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
-	c.Assert(err, IsNil)
-	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
-	c.Assert(err, IsNil)
-	checkAndFalse := false
-	// previous session got signaled by closing its channel
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
+	c.Assert(err, IsNil)
+	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s2")
+	c.Assert(err, IsNil)
+	// previous session got signaled by sending nil on its channel
+	var sentinel broker.Exchange
+	got := false
 	select {
-	case _, ok := <-sess1.SessionChannel():
-		checkAndFalse = ok == false
-	default:
+	case sentinel = <-sess1.SessionChannel():
+		got = true
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get sentinel")
 	}
-	c.Check(checkAndFalse, Equals, true)
+	c.Check(got, Equals, true)
+	c.Check(sentinel, IsNil)
 	c.Assert(s.RevealSession(b, "dev-1"), Equals, sess2)
 	b.Unregister(sess1)
 	// just to make sure the unregister was processed
-	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""})
+	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""}, "s3")
 	c.Assert(err, IsNil)
 	c.Check(s.RevealSession(b, "dev-1"), Equals, sess2)
 }
@@ -187,9 +192,9 @@
 	b := s.MakeBroker(sto, testBrokerConfig, nil)
 	b.Start()
 	defer b.Stop()
-	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
-	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-2"})
+	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-2"}, "s2")
 	c.Assert(err, IsNil)
 	// add notification to channel *after* the registrations
 	muchLater := time.Now().Add(10 * time.Minute)
@@ -200,10 +205,10 @@
 		c.Fatal("taking too long to get broadcast exchange")
 	case exchg1 := <-sess1.SessionChannel():
 		c.Check(s.RevealBroadcastExchange(exchg1), DeepEquals, &broker.BroadcastExchange{
-			ChanId:               store.SystemInternalChannelId,
-			TopLevel:             1,
-			NotificationPayloads: []json.RawMessage{notification1},
-			Decoded:              []map[string]interface{}{decoded1},
+			ChanId:        store.SystemInternalChannelId,
+			TopLevel:      1,
+			Notifications: help.Ns(notification1),
+			Decoded:       []map[string]interface{}{decoded1},
 		})
 	}
 	select {
@@ -211,10 +216,10 @@
 		c.Fatal("taking too long to get broadcast exchange")
 	case exchg2 := <-sess2.SessionChannel():
 		c.Check(s.RevealBroadcastExchange(exchg2), DeepEquals, &broker.BroadcastExchange{
-			ChanId:               store.SystemInternalChannelId,
-			TopLevel:             1,
-			NotificationPayloads: []json.RawMessage{notification1},
-			Decoded:              []map[string]interface{}{decoded1},
+			ChanId:        store.SystemInternalChannelId,
+			TopLevel:      1,
+			Notifications: help.Ns(notification1),
+			Decoded:       []map[string]interface{}{decoded1},
 		})
 	}
 }
@@ -224,7 +229,7 @@
 	countdownToFail int
 }
 
-func (sto *testFailingStore) GetChannelSnapshot(chanId store.InternalChannelId) (int64, []json.RawMessage, error) {
+func (sto *testFailingStore) GetChannelSnapshot(chanId store.InternalChannelId) (int64, []protocol.Notification, error) {
 	if sto.countdownToFail == 0 {
 		return 0, nil, errors.New("get channel snapshot fail")
 	}
@@ -232,6 +237,10 @@
 	return 0, nil, nil
 }
 
+func (sto *testFailingStore) DropByMsgId(chanId store.InternalChannelId, targets []protocol.Notification) error {
+	return errors.New("drop fail")
+}
+
 func (s *CommonBrokerSuite) TestBroadcastFail(c *C) {
 	logged := make(chan bool, 1)
 	s.testlog.SetLogEventCb(func(string) {
@@ -241,7 +250,7 @@
 	b := s.MakeBroker(sto, testBrokerConfig, s.testlog)
 	b.Start()
 	defer b.Stop()
-	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"}, "s1")
 	c.Assert(err, IsNil)
 	b.Broadcast(store.SystemInternalChannelId)
 	select {
@@ -249,5 +258,83 @@
 		c.Fatal("taking too long to log error")
 	case <-logged:
 	}
-	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful broadcast, get channel snapshot for 0: get channel snapshot fail\n")
+	c.Check(s.testlog.Captured(), Matches, "ERROR.*: get channel snapshot fail\n")
+}
+
+func (s *CommonBrokerSuite) TestUnicast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M1"}`)
+	notification2 := json.RawMessage(`{"m": "M2"}`)
+	chanId1 := store.UnicastInternalChannelId("dev1", "dev1")
+	chanId2 := store.UnicastInternalChannelId("dev2", "dev2")
+	b := s.MakeBroker(sto, testBrokerConfig, nil)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev1"}, "s1")
+	c.Assert(err, IsNil)
+	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev2"}, "s2")
+	c.Assert(err, IsNil)
+	// add notification to channel *after* the registrations
+	muchLater := time.Now().Add(10 * time.Minute)
+	sto.AppendToUnicastChannel(chanId1, "app1", notification1, "msg1", muchLater)
+	sto.AppendToUnicastChannel(chanId2, "app2", notification2, "msg2", muchLater)
+	b.Unicast(chanId2, chanId1)
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get unicast exchange")
+	case exchg1 := <-sess1.SessionChannel():
+		u1 := s.RevealUnicastExchange(exchg1)
+		c.Check(u1.ChanId, Equals, chanId1)
+	}
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get unicast exchange")
+	case exchg2 := <-sess2.SessionChannel():
+		u2 := s.RevealUnicastExchange(exchg2)
+		c.Check(u2.ChanId, Equals, chanId2)
+	}
+}
+
+func (s *CommonBrokerSuite) TestGetAndDrop(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M1"}`)
+	chanId1 := store.UnicastInternalChannelId("dev3", "dev3")
+	b := s.MakeBroker(sto, testBrokerConfig, nil)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev3"}, "s1")
+	c.Assert(err, IsNil)
+	muchLater := time.Now().Add(10 * time.Minute)
+	sto.AppendToUnicastChannel(chanId1, "app1", notification1, "msg1", muchLater)
+	_, expected, err := sto.GetChannelSnapshot(chanId1)
+	c.Assert(err, IsNil)
+	_, notifs, err := sess1.Get(chanId1, false)
+	c.Check(notifs, HasLen, 1)
+	c.Check(notifs, DeepEquals, expected)
+	err = sess1.DropByMsgId(chanId1, notifs)
+	c.Assert(err, IsNil)
+	_, notifs, err = sess1.Get(chanId1, true)
+	c.Check(notifs, HasLen, 0)
+	_, expected, err = sto.GetChannelSnapshot(chanId1)
+	c.Assert(err, IsNil)
+	c.Check(expected, HasLen, 0)
+
+}
+
+func (s *CommonBrokerSuite) TestGetAndDropErrors(c *C) {
+	chanId1 := store.UnicastInternalChannelId("dev3", "dev3")
+	sto := &testFailingStore{countdownToFail: 1}
+	b := s.MakeBroker(sto, testBrokerConfig, s.testlog)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev3"}, "s1")
+	c.Assert(err, IsNil)
+	_, _, err = sess1.Get(chanId1, false)
+	c.Assert(err, ErrorMatches, "get channel snapshot fail")
+	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful, get channel snapshot for Udev3:dev3 \\(cachedOk=false\\): get channel snapshot fail\n")
+	s.testlog.ResetCapture()
+
+	err = sess1.DropByMsgId(chanId1, nil)
+	c.Assert(err, ErrorMatches, "drop fail")
+	c.Check(s.testlog.Captured(), Matches, "ERROR unsuccessful, drop from channel Udev3:dev3: drop fail\n")
 }

=== modified file 'server/session/session.go'
--- server/session/session.go	2014-04-11 08:47:18 +0000
+++ server/session/session.go	2014-05-12 14:11:46 +0000
@@ -18,6 +18,7 @@
 package session
 
 import (
+	"errors"
 	"net"
 	"time"
 
@@ -35,7 +36,7 @@
 }
 
 // sessionStart manages the start of the protocol session.
-func sessionStart(proto protocol.Protocol, brkr broker.Broker, cfg SessionConfig) (broker.BrokerSession, error) {
+func sessionStart(proto protocol.Protocol, brkr broker.Broker, cfg SessionConfig, sessionId string) (broker.BrokerSession, error) {
 	var connMsg protocol.ConnectMsg
 	proto.SetDeadline(time.Now().Add(cfg.ExchangeTimeout()))
 	err := proto.ReadMessage(&connMsg)
@@ -52,9 +53,11 @@
 	if err != nil {
 		return nil, err
 	}
-	return brkr.Register(&connMsg)
+	return brkr.Register(&connMsg, sessionId)
 }
 
+var errOneway = errors.New("oneway")
+
 // exchange writes outMsg message, reads answer in inMsg
 func exchange(proto protocol.Protocol, outMsg, inMsg interface{}, exchangeTimeout time.Duration) error {
 	proto.SetDeadline(time.Now().Add(exchangeTimeout))
@@ -62,7 +65,10 @@
 	if err != nil {
 		return err
 	}
-	if inMsg == nil { // no answer expected, breaking connection
+	if inMsg == nil { // no answer expected
+		if outMsg.(protocol.OnewayMsg).OnewayContinue() {
+			return errOneway
+		}
 		return &broker.ErrAbort{"session broken for reason"}
 	}
 	err = proto.ReadMessage(inMsg)
@@ -78,6 +84,10 @@
 	exchangeTimeout := cfg.ExchangeTimeout()
 	pingTimer := time.NewTimer(pingInterval)
 	intervalStart := time.Now()
+	pingTimerReset := func() {
+		pingTimer.Reset(pingInterval)
+		intervalStart = time.Now()
+	}
 	ch := sess.SessionChannel()
 Loop:
 	for {
@@ -93,16 +103,15 @@
 			if pongMsg.Type != "pong" {
 				return &broker.ErrAbort{"expected PONG message"}
 			}
-			pingTimer.Reset(pingInterval)
-		case exchg, ok := <-ch:
+			pingTimerReset()
+		case exchg := <-ch:
 			pingTimer.Stop()
-			if !ok {
+			if exchg == nil {
 				return &broker.ErrAbort{"terminated"}
 			}
 			outMsg, inMsg, err := exchg.Prepare(sess)
 			if err == broker.ErrNop { // nothing to do
-				pingTimer.Reset(pingInterval)
-				intervalStart = time.Now()
+				pingTimerReset()
 				continue Loop
 			}
 			if err != nil {
@@ -111,12 +120,15 @@
 			for {
 				done := outMsg.Split()
 				err = exchange(proto, outMsg, inMsg, exchangeTimeout)
+				if err == errOneway {
+					pingTimerReset()
+					continue Loop
+				}
 				if err != nil {
 					return err
 				}
 				if done {
-					pingTimer.Reset(pingInterval)
-					intervalStart = time.Now()
+					pingTimerReset()
 				}
 				err = exchg.Acked(sess, done)
 				if err != nil {
@@ -142,7 +154,7 @@
 		return track.End(&broker.ErrAbort{"unexpected wire format version"})
 	}
 	proto := protocol.NewProtocol0(conn)
-	sess, err := sessionStart(proto, brkr, cfg)
+	sess, err := sessionStart(proto, brkr, cfg, track.SessionId())
 	if err != nil {
 		return track.End(err)
 	}

=== modified file 'server/session/session_test.go'
--- server/session/session_test.go	2014-04-11 08:47:18 +0000
+++ server/session/session_test.go	2014-05-12 14:11:46 +0000
@@ -130,8 +130,8 @@
 	return &testBroker{registration: make(chan interface{}, 2)}
 }
 
-func (tb *testBroker) Register(connect *protocol.ConnectMsg) (broker.BrokerSession, error) {
-	tb.registration <- "register " + connect.DeviceId
+func (tb *testBroker) Register(connect *protocol.ConnectMsg, sessionId string) (broker.BrokerSession, error) {
+	tb.registration <- fmt.Sprintf("register %s %s", connect.DeviceId, sessionId)
 	return &testing.TestBrokerSession{DeviceId: connect.DeviceId}, tb.err
 }
 
@@ -148,7 +148,7 @@
 	brkr := newTestBroker()
 	go func() {
 		var err error
-		sess, err = sessionStart(tp, brkr, cfg10msPingInterval5msExchangeTout)
+		sess, err = sessionStart(tp, brkr, cfg10msPingInterval5msExchangeTout, "s1")
 		errCh <- err
 	}()
 	c.Check(takeNext(down), Equals, "deadline 5ms")
@@ -160,7 +160,7 @@
 	up <- nil // no write error
 	err := <-errCh
 	c.Check(err, IsNil)
-	c.Check(takeNext(brkr.registration), Equals, "register dev-1")
+	c.Check(takeNext(brkr.registration), Equals, "register dev-1 s1")
 	c.Check(sess.DeviceIdentifier(), Equals, "dev-1")
 }
 
@@ -175,7 +175,7 @@
 	brkr.err = errRegister
 	go func() {
 		var err error
-		sess, err = sessionStart(tp, brkr, cfg10msPingInterval5msExchangeTout)
+		sess, err = sessionStart(tp, brkr, cfg10msPingInterval5msExchangeTout, "s2")
 		errCh <- err
 	}()
 	up <- protocol.ConnectMsg{Type: "connect", ClientVer: "1", DeviceId: "dev-1"}
@@ -190,7 +190,7 @@
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	up <- io.ErrUnexpectedEOF
-	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout)
+	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout, "s3")
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
 }
 
@@ -200,7 +200,7 @@
 	tp := &testProtocol{up, down}
 	up <- protocol.ConnectMsg{Type: "connect"}
 	up <- io.ErrUnexpectedEOF
-	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout)
+	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout, "s4")
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
 	// sanity
 	c.Check(takeNext(down), Matches, "deadline.*")
@@ -212,7 +212,7 @@
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	up <- protocol.ConnectMsg{Type: "what"}
-	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout)
+	_, err := sessionStart(tp, nil, cfg10msPingInterval5msExchangeTout, "s5")
 	c.Check(err, DeepEquals, &broker.ErrAbort{"expected CONNECT message"})
 }
 
@@ -222,14 +222,14 @@
 }
 
 func (s *sessionSuite) TestSessionLoop(c *C) {
-	nopTrack := NewTracker(s.testlog)
+	track := &testTracker{NewTracker(s.testlog), make(chan interface{}, 2)}
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	sess := &testing.TestBrokerSession{}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, track)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
@@ -241,6 +241,9 @@
 	up <- io.ErrUnexpectedEOF
 	err := <-errCh
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
+	c.Check(track.interval, HasLen, 2)
+	c.Check((<-track.interval).(time.Duration) <= 8*time.Millisecond, Equals, true)
+	c.Check((<-track.interval).(time.Duration) <= 8*time.Millisecond, Equals, true)
 }
 
 func (s *sessionSuite) TestSessionLoopWriteError(c *C) {
@@ -357,7 +360,7 @@
 	go func() {
 		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
-	close(exchanges)
+	exchanges <- nil
 	err := <-errCh
 	c.Check(err, DeepEquals, &broker.ErrAbort{"terminated"})
 }
@@ -477,18 +480,44 @@
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	exchanges := make(chan broker.Exchange, 1)
-	exchanges <- &broker.ConnBrokenExchange{"REASON"}
+	msg := &protocol.ConnBrokenMsg{"connbroken", "BREASON"}
+	exchanges <- &broker.ConnMetaExchange{msg}
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
 		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
-	c.Check(takeNext(down), DeepEquals, protocol.ConnBrokenMsg{"connbroken", "REASON"})
+	c.Check(takeNext(down), DeepEquals, protocol.ConnBrokenMsg{"connbroken", "BREASON"})
 	up <- nil // no write error
 	err := <-errCh
 	c.Check(err, DeepEquals, &broker.ErrAbort{"session broken for reason"})
 }
 
+func (s *sessionSuite) TestSessionLoopConnWarnExchange(c *C) {
+	nopTrack := NewTracker(s.testlog)
+	errCh := make(chan error, 1)
+	up := make(chan interface{}, 5)
+	down := make(chan interface{}, 5)
+	tp := &testProtocol{up, down}
+	exchanges := make(chan broker.Exchange, 1)
+	msg := &protocol.ConnWarnMsg{"connwarn", "WREASON"}
+	exchanges <- &broker.ConnMetaExchange{msg}
+	sess := &testing.TestBrokerSession{Exchanges: exchanges}
+	go func() {
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
+	}()
+	c.Check(takeNext(down), Equals, "deadline 2ms")
+	c.Check(takeNext(down), DeepEquals, protocol.ConnWarnMsg{"connwarn", "WREASON"})
+	up <- nil // no write error
+	// session continues
+	c.Check(takeNext(down), Equals, "deadline 2ms")
+	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
+	up <- nil // no write error
+	up <- io.EOF
+	err := <-errCh
+	c.Check(err, Equals, io.EOF)
+}
+
 type testTracker struct {
 	SessionTracker
 	interval chan interface{}
@@ -593,7 +622,7 @@
 	msg, err = downStream.ReadBytes(byte('}'))
 	c.Check(err, IsNil)
 	c.Check(msg, DeepEquals, []byte("\x00\x0c{\"T\":\"ping\"}"))
-	c.Check(takeNext(brkr.registration), Equals, "register DEV")
+	c.Check(takeNext(brkr.registration), Equals, "register DEV "+track.SessionId())
 	c.Check(len(brkr.registration), Equals, 0) // not yet unregistered
 	cli.Close()
 	err = <-errCh

=== modified file 'server/session/tracker.go'
--- server/session/tracker.go	2014-02-10 23:19:08 +0000
+++ server/session/tracker.go	2014-05-12 14:11:46 +0000
@@ -17,6 +17,7 @@
 package session
 
 import (
+	"fmt"
 	"net"
 	"time"
 
@@ -29,6 +30,8 @@
 	logger.Logger
 	// Session got started.
 	Start(WithRemoteAddr)
+	// SessionId
+	SessionId() string
 	// Session got registered with broker as sess BrokerSession.
 	Registered(sess broker.BrokerSession)
 	// Report effective elapsed ping interval.
@@ -47,7 +50,7 @@
 // Tracker implements SessionTracker simply.
 type tracker struct {
 	logger.Logger
-	sessionId int64 // xxx use timeuuid later
+	sessionId string
 }
 
 func NewTracker(logger logger.Logger) SessionTracker {
@@ -55,18 +58,22 @@
 }
 
 func (trk *tracker) Start(conn WithRemoteAddr) {
-	trk.sessionId = time.Now().UnixNano() - sessionsEpoch
-	trk.Debugf("session(%x) connected %v", trk.sessionId, conn.RemoteAddr())
+	trk.sessionId = fmt.Sprintf("%x", time.Now().UnixNano()-sessionsEpoch)
+	trk.Debugf("session(%s) connected %v", trk.sessionId, conn.RemoteAddr())
+}
+
+func (trk *tracker) SessionId() string {
+	return trk.sessionId
 }
 
 func (trk *tracker) Registered(sess broker.BrokerSession) {
-	trk.Infof("session(%x) registered %v", trk.sessionId, sess.DeviceIdentifier())
+	trk.Infof("session(%s) registered %v", trk.sessionId, sess.DeviceIdentifier())
 }
 
 func (trk *tracker) EffectivePingInterval(time.Duration) {
 }
 
 func (trk *tracker) End(err error) error {
-	trk.Debugf("session(%x) ended with: %v", trk.sessionId, err)
+	trk.Debugf("session(%s) ended with: %v", trk.sessionId, err)
 	return err
 }

=== modified file 'server/session/tracker_test.go'
--- server/session/tracker_test.go	2014-02-10 23:19:08 +0000
+++ server/session/tracker_test.go	2014-05-12 14:11:46 +0000
@@ -46,8 +46,8 @@
 func (s *trackerSuite) TestSessionTrackStart(c *C) {
 	track := NewTracker(s.testlog)
 	track.Start(&testRemoteAddrable{})
-	c.Check(track.(*tracker).sessionId, Not(Equals), 0)
-	regExpected := fmt.Sprintf(`DEBUG session\(%x\) connected 127\.0\.0\.1:9999\n`, track.(*tracker).sessionId)
+	c.Check(track.SessionId(), Not(Equals), "")
+	regExpected := fmt.Sprintf(`DEBUG session\(%s\) connected 127\.0\.0\.1:9999\n`, track.SessionId())
 	c.Check(s.testlog.Captured(), Matches, regExpected)
 }
 
@@ -55,7 +55,7 @@
 	track := NewTracker(s.testlog)
 	track.Start(&testRemoteAddrable{})
 	track.Registered(&testing.TestBrokerSession{DeviceId: "DEV-ID"})
-	regExpected := fmt.Sprintf(`.*connected.*\nINFO session\(%x\) registered DEV-ID\n`, track.(*tracker).sessionId)
+	regExpected := fmt.Sprintf(`.*connected.*\nINFO session\(%s\) registered DEV-ID\n`, track.SessionId())
 	c.Check(s.testlog.Captured(), Matches, regExpected)
 }
 
@@ -63,6 +63,6 @@
 	track := NewTracker(s.testlog)
 	track.Start(&testRemoteAddrable{})
 	track.End(&broker.ErrAbort{})
-	regExpected := fmt.Sprintf(`.*connected.*\nDEBUG session\(%x\) ended with: session aborted \(\)\n`, track.(*tracker).sessionId)
+	regExpected := fmt.Sprintf(`.*connected.*\nDEBUG session\(%s\) ended with: session aborted \(\)\n`, track.SessionId())
 	c.Check(s.testlog.Captured(), Matches, regExpected)
 }

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

=== modified file 'server/store/inmemory_test.go'
--- server/store/inmemory_test.go	2014-02-14 12:38:38 +0000
+++ server/store/inmemory_test.go	2014-05-12 14:11:46 +0000
@@ -21,6 +21,9 @@
 	"time"
 
 	. "launchpad.net/gocheck"
+
+	"launchpad.net/ubuntu-push/protocol"
+	help "launchpad.net/ubuntu-push/testing"
 )
 
 type inMemorySuite struct{}
@@ -45,7 +48,7 @@
 	top, res, err := sto.GetChannelSnapshot(SystemInternalChannelId)
 	c.Assert(err, IsNil)
 	c.Check(top, Equals, int64(0))
-	c.Check(res, DeepEquals, []json.RawMessage(nil))
+	c.Check(res, DeepEquals, []protocol.Notification(nil))
 }
 
 func (s *inMemorySuite) TestAppendToChannelAndGetChannelSnapshot(c *C) {
@@ -61,7 +64,33 @@
 	top, res, err := sto.GetChannelSnapshot(SystemInternalChannelId)
 	c.Assert(err, IsNil)
 	c.Check(top, Equals, int64(2))
-	c.Check(res, DeepEquals, []json.RawMessage{notification1, notification2})
+	c.Check(res, DeepEquals, help.Ns(notification1, notification2))
+}
+
+func (s *inMemorySuite) TestAppendToUnicastChannelAndGetChannelSnapshot(c *C) {
+	sto := NewInMemoryPendingStore()
+
+	chanId := UnicastInternalChannelId("user", "dev1")
+	notification1 := json.RawMessage(`{"a":1}`)
+	notification2 := json.RawMessage(`{"b":2}`)
+	app1 := "app1"
+	app2 := "app2"
+	msg1 := "msg1"
+	msg2 := "msg2"
+
+	muchLater := time.Now().Add(time.Minute)
+
+	err := sto.AppendToUnicastChannel(chanId, app1, notification1, msg1, muchLater)
+	c.Assert(err, IsNil)
+	err = sto.AppendToUnicastChannel(chanId, app2, notification2, msg2, muchLater)
+	c.Assert(err, IsNil)
+	top, res, err := sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+	c.Check(res, DeepEquals, []protocol.Notification{
+		protocol.Notification{Payload: notification1, AppId: app1, MsgId: msg1},
+		protocol.Notification{Payload: notification2, AppId: app2, MsgId: msg2},
+	})
+	c.Check(top, Equals, int64(0))
 }
 
 func (s *inMemorySuite) TestAppendToChannelAndGetChannelSnapshotWithExpiration(c *C) {
@@ -81,5 +110,45 @@
 	top, res, err := sto.GetChannelSnapshot(SystemInternalChannelId)
 	c.Assert(err, IsNil)
 	c.Check(top, Equals, int64(2))
-	c.Check(res, DeepEquals, []json.RawMessage{notification1})
+	c.Check(res, DeepEquals, help.Ns(notification1))
+}
+
+func (s *inMemorySuite) TestDropByMsgId(c *C) {
+	sto := NewInMemoryPendingStore()
+
+	chanId := UnicastInternalChannelId("user", "dev2")
+
+	// nothing to do is fine
+	err := sto.DropByMsgId(chanId, nil)
+	c.Assert(err, IsNil)
+
+	notification1 := json.RawMessage(`{"a":1}`)
+	notification2 := json.RawMessage(`{"b":2}`)
+	notification3 := json.RawMessage(`{"a":2}`)
+	app1 := "app1"
+	app2 := "app2"
+	msg1 := "msg1"
+	msg2 := "msg2"
+	msg3 := "msg3"
+
+	muchLater := time.Now().Add(time.Minute)
+
+	err = sto.AppendToUnicastChannel(chanId, app1, notification1, msg1, muchLater)
+	c.Assert(err, IsNil)
+	err = sto.AppendToUnicastChannel(chanId, app2, notification2, msg2, muchLater)
+	c.Assert(err, IsNil)
+	err = sto.AppendToUnicastChannel(chanId, app1, notification3, msg3, muchLater)
+	c.Assert(err, IsNil)
+	_, res, err := sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+
+	err = sto.DropByMsgId(chanId, res[:2])
+	c.Assert(err, IsNil)
+
+	_, res, err = sto.GetChannelSnapshot(chanId)
+	c.Assert(err, IsNil)
+	c.Check(res, HasLen, 1)
+	c.Check(res, DeepEquals, []protocol.Notification{
+		protocol.Notification{Payload: notification3, AppId: app1, MsgId: msg3},
+	})
 }

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

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

=== added directory 'signing-helper'
=== added file 'signing-helper/CMakeLists.txt'
--- signing-helper/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ signing-helper/CMakeLists.txt	2014-05-12 14:11:46 +0000
@@ -0,0 +1,39 @@
+cmake_minimum_required(VERSION 2.8)
+SET (EXAMPLES_TARGET ubuntuoneauth-examples)
+
+SET (SIGNING_EXE "signing-helper")
+
+find_package (PkgConfig REQUIRED)
+pkg_check_modules(UBUNTUONE REQUIRED ubuntuoneauth-2.0)
+add_definitions(${UBUNTUONE_CFLAGS} ${UBUNTUONE_CFLAGS_OTHER})
+
+
+# Qt5 bits
+SET (CMAKE_INCLUDE_CURRENT_DIR ON)
+SET (CMAKE_AUTOMOC ON)
+find_package(Qt5Core REQUIRED)
+
+FILE (GLOB SIGNING_SOURCES signing*.cpp)
+FILE (GLOB SIGNING_HEADERS signing*.h)
+
+add_executable (${SIGNING_EXE}
+  ${SIGNING_SOURCES}
+  ${SIGNING_HEADERS})
+qt5_use_modules (${SIGNING_EXE} DBus Network)
+
+target_link_libraries (${SIGNING_EXE}
+  ${UBUNTUONE_LDFLAGS})
+
+
+
+add_custom_target(examples-valgrind
+    COMMAND "valgrind --tool=memcheck ${CMAKE_CURRENT_BINARY_DIR}/${SIGNING_EXE}"
+    DEPENDS ${SIGNING_EXE}
+)
+
+add_custom_target(examples-valgrind-leaks
+    COMMAND "valgrind --tool=memcheck --track-origins=yes --num-callers=40 --leak-resolution=high --leak-check=full ${CMAKE_CURRENT_BINARY_DIR}/${SIGNING_EXE}"
+    DEPENDS ${SIGNING_EXE}
+)
+
+INSTALL_TARGETS( "lib/ubuntu-push-client/" ${SIGNING_EXE})

=== added file 'signing-helper/signing-helper.cpp'
--- signing-helper/signing-helper.cpp	1970-01-01 00:00:00 +0000
+++ signing-helper/signing-helper.cpp	2014-05-12 14:11:46 +0000
@@ -0,0 +1,97 @@
+/*
+ * Copyright (C) 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/>.
+ *
+ * In addition, as a special exception, the copyright holders give
+ * permission to link the code of portions of this program with the
+ * OpenSSL library under certain conditions as described in each
+ * individual source file, and distribute linked combinations
+ * including the two.
+ * You must obey the GNU General Public License in all respects
+ * for all of the code used other than OpenSSL.  If you modify
+ * file(s) with this exception, you may extend this exception to your
+ * version of the file(s), but you are not obligated to do so.  If you
+ * do not wish to do so, delete this exception statement from your
+ * version.  If you delete this exception statement from all source
+ * files in the program, then also delete it here.
+ */
+
+#include <iostream>
+#include <QCoreApplication>
+#include <QDebug>
+#include <QObject>
+#include <QString>
+#include <QTimer>
+
+#include "ssoservice.h"
+#include "token.h"
+
+#include "signing.h"
+
+namespace UbuntuOne {
+
+    SigningExample::SigningExample(QObject *parent, QString url) :
+        QObject(parent)
+    {
+        QObject::connect(&service, SIGNAL(credentialsFound(const Token&)),
+                         this, SLOT(handleCredentialsFound(Token)));
+        QObject::connect(&service, SIGNAL(credentialsNotFound()),
+                         this, SLOT(handleCredentialsNotFound()));
+        this->url = url;
+
+    }
+
+    SigningExample::~SigningExample(){
+    }
+
+    void SigningExample::doExample()
+    {
+        service.getCredentials();
+    }
+
+    void SigningExample::handleCredentialsFound(Token token)
+    {
+        qDebug() << "Credentials found, signing url.";
+
+        QString authHeader = token.signUrl(this->url, QStringLiteral("GET"), true);
+
+        std::cout << authHeader.toStdString() << "\n";
+        QCoreApplication::instance()->exit(0);
+
+    }
+
+    void SigningExample::handleCredentialsNotFound()
+    {
+        qDebug() << "No credentials were found.";
+        QCoreApplication::instance()->exit(1);
+    }
+
+
+} // namespace UbuntuOne
+
+
+int main(int argc, char *argv[])
+{
+    QCoreApplication a(argc, argv);
+
+    UbuntuOne::SigningExample *example = new UbuntuOne::SigningExample(&a);
+
+    QObject::connect(example, SIGNAL(finished()), &a, SLOT(quit()));
+
+    QTimer::singleShot(0, example, SLOT(doExample()));
+
+    return a.exec();
+}
+
+

=== added file 'signing-helper/signing.h'
--- signing-helper/signing.h	1970-01-01 00:00:00 +0000
+++ signing-helper/signing.h	2014-05-12 14:11:46 +0000
@@ -0,0 +1,76 @@
+/*
+ * Copyright (C) 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/>.
+ *
+ * In addition, as a special exception, the copyright holders give
+ * permission to link the code of portions of this program with the
+ * OpenSSL library under certain conditions as described in each
+ * individual source file, and distribute linked combinations
+ * including the two.
+ * You must obey the GNU General Public License in all respects
+ * for all of the code used other than OpenSSL.  If you modify
+ * file(s) with this exception, you may extend this exception to your
+ * version of the file(s), but you are not obligated to do so.  If you
+ * do not wish to do so, delete this exception statement from your
+ * version.  If you delete this exception statement from all source
+ * files in the program, then also delete it here.
+ */
+
+#ifndef _SIGNING_H_
+#define _SIGNING_H_
+
+#include <QDebug>
+#include <QNetworkReply>
+#include <QObject>
+#include <QString>
+
+#include "ssoservice.h"
+#include "token.h"
+#include "requests.h"
+#include "errormessages.h"
+
+namespace UbuntuOne {
+
+class SigningExample : public QObject
+{
+    Q_OBJECT
+
+public:
+
+    explicit SigningExample(QObject *parent = 0, QString url="https://one.ubuntu.com/api/account/");
+    ~SigningExample();
+
+public slots:
+
+    void doExample();
+
+signals:
+
+    void finished();
+
+private slots:
+
+    void handleCredentialsFound(Token token);
+    void handleCredentialsNotFound();
+
+private:
+
+    SSOService service;
+    QNetworkAccessManager nam;
+    QString url;
+
+};
+
+}
+#endif /* _SIGNING_H_ */

=== modified file 'testing/helpers.go'
--- testing/helpers.go	2014-02-21 16:04:44 +0000
+++ testing/helpers.go	2014-05-12 14:11:46 +0000
@@ -18,6 +18,7 @@
 package testing
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -26,6 +27,7 @@
 	"sync"
 
 	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
 )
 
 type captureHelper struct {
@@ -122,3 +124,12 @@
 	}
 	return filepath.Join(dir, relativePath)
 }
+
+// Ns makes a []Notification from just payloads.
+func Ns(payloads ...json.RawMessage) []protocol.Notification {
+	res := make([]protocol.Notification, len(payloads))
+	for i := 0; i < len(payloads); i++ {
+		res[i].Payload = payloads[i]
+	}
+	return res
+}

=== modified file 'ubuntu-push-client.go'
--- ubuntu-push-client.go	2014-04-15 11:27:38 +0000
+++ ubuntu-push-client.go	2014-05-12 14:11:46 +0000
@@ -58,6 +58,7 @@
 	if err != nil {
 		log.Fatalf("unable to open the levels database: %v", err)
 	}
+
 	cli := client.NewPushClient(cfgFname, lvlFname)
 	err = cli.Start()
 	if err != nil {