Merge lp:ubuntu-push/automatic into lp:ubuntu-push

[John Lenton]

Merging automatic.

=== modified file 'client/client.go'
--- client/client.go	2014-04-03 20:56:25 +0000
+++ client/client.go	2014-04-11 18:18:05 +0000
@@ -81,6 +81,8 @@
 	sessionConnectedCh chan uint32
 }
 
+var ACTION_ID_SNOWFLAKE = "::ubuntu-push-client::"
+
 // Creates a new Ubuntu Push Notifications client-side daemon that will use
 // the given configuration file.
 func NewPushClient(configPath string, leveldbPath string) *PushClient {
@@ -278,7 +280,7 @@
 	if !client.filterNotification(msg) {
 		return nil
 	}
-	action_id := "dummy_id"
+	action_id := ACTION_ID_SNOWFLAKE
 	a := []string{action_id, "Go get it!"} // action value not visible on the phone
 	h := map[string]*dbus.Variant{"x-canonical-switch-to-application": &dbus.Variant{true}}
 	nots := notifications.Raw(client.notificationsEndp, client.log)
@@ -305,20 +307,23 @@
 }
 
 // handleClick deals with the user clicking a notification
-func (client *PushClient) handleClick() error {
+func (client *PushClient) handleClick(action_id string) error {
+	if action_id != ACTION_ID_SNOWFLAKE {
+		return nil
+	}
 	// it doesn't get much simpler...
 	urld := urldispatcher.New(client.urlDispatcherEndp, client.log)
 	return urld.DispatchURL("settings:///system/system-update")
 }
 
 // doLoop connects events with their handlers
-func (client *PushClient) doLoop(connhandler func(bool), clickhandler func() error, notifhandler func(*session.Notification) error, errhandler func(error)) {
+func (client *PushClient) doLoop(connhandler func(bool), clickhandler func(string) error, notifhandler func(*session.Notification) error, errhandler func(error)) {
 	for {
 		select {
 		case state := <-client.connCh:
 			connhandler(state)
-		case <-client.actionsCh:
-			clickhandler()
+		case action := <-client.actionsCh:
+			clickhandler(action.ActionId)
 		case msg := <-client.session.MsgCh:
 			notifhandler(msg)
 		case err := <-client.session.ErrCh:

=== modified file 'client/client_test.go'
--- client/client_test.go	2014-04-03 20:56:25 +0000
+++ client/client_test.go	2014-04-11 18:18:05 +0000
@@ -588,9 +588,15 @@
 	cli.log = cs.log
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(true))
 	cli.urlDispatcherEndp = endp
-	c.Check(cli.handleClick(), IsNil)
+	// check we don't fail on something random
+	c.Check(cli.handleClick("something random"), IsNil)
+	// ... but we don't send anything either
+	args := testibus.GetCallArgs(endp)
+	c.Assert(args, HasLen, 0)
+	// check we worked with the right action id
+	c.Check(cli.handleClick(ACTION_ID_SNOWFLAKE), IsNil)
 	// check we sent the notification
-	args := testibus.GetCallArgs(endp)
+	args = testibus.GetCallArgs(endp)
 	c.Assert(args, HasLen, 1)
 	c.Check(args[0].Member, Equals, "DispatchURL")
 	c.Check(args[0].Args, DeepEquals, []interface{}{"settings:///system/system-update"})
@@ -609,7 +615,7 @@
 	c.Assert(cli.initSession(), IsNil)
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) { ch <- true }, func() error { return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
+	go cli.doLoop(func(bool) { ch <- true }, func(_ string) error { return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -623,7 +629,7 @@
 	cli.actionsCh = aCh
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func() error { ch <- true; return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
+	go cli.doLoop(func(bool) {}, func(_ string) error { ch <- true; return nil }, func(_ *session.Notification) error { return nil }, func(error) {})
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -636,7 +642,7 @@
 	cli.session.MsgCh <- &session.Notification{}
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func() error { return nil }, func(_ *session.Notification) error { ch <- true; return nil }, func(error) {})
+	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.Notification) error { ch <- true; return nil }, func(error) {})
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -649,7 +655,7 @@
 	cli.session.ErrCh <- nil
 
 	ch := make(chan bool, 1)
-	go cli.doLoop(func(bool) {}, func() error { return nil }, func(_ *session.Notification) error { return nil }, func(error) { ch <- true })
+	go cli.doLoop(func(bool) {}, func(_ string) error { return nil }, func(_ *session.Notification) error { return nil }, func(error) { ch <- true })
 	c.Check(takeNextBool(ch), Equals, true)
 }
 
@@ -716,7 +722,7 @@
 	c.Check(cs.log.Captured(), Matches, "(?ms).*Session connected after 42 attempts$")
 
 	//  * actionsCh to the click handler/url dispatcher
-	aCh <- notifications.RawActionReply{}
+	aCh <- notifications.RawActionReply{ActionId: ACTION_ID_SNOWFLAKE}
 	tick()
 	uargs := testibus.GetCallArgs(cli.urlDispatcherEndp)
 	c.Assert(uargs, HasLen, 1)

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-04-04 12:28:40 +0000
+++ client/session/session.go	2014-04-11 18:18:05 +0000
@@ -49,6 +49,7 @@
 	Type string `json:"T"`
 	protocol.BroadcastMsg
 	protocol.NotificationsMsg
+	protocol.ConnBrokenMsg
 }
 
 // parseServerAddrSpec recognizes whether spec is a HTTP URL to get
@@ -176,7 +177,7 @@
 // getHosts sets deliveryHosts possibly querying a remote endpoint
 func (sess *ClientSession) getHosts() error {
 	if sess.getHost != nil {
-		if sess.timeSince(sess.deliveryHostsTimestamp) < sess.HostsCachingExpiryTime {
+		if sess.deliveryHosts != nil && sess.timeSince(sess.deliveryHostsTimestamp) < sess.HostsCachingExpiryTime {
 			return nil
 		}
 		hosts, err := sess.getHost.Get()
@@ -193,6 +194,10 @@
 	return nil
 }
 
+func (sess *ClientSession) resetHosts() {
+	sess.deliveryHosts = nil
+}
+
 // startConnectionAttempt/nextHostToTry help connect iterating over candidate hosts
 
 func (sess *ClientSession) startConnectionAttempt() {
@@ -338,6 +343,19 @@
 	return nil
 }
 
+// handle "connbroken" messages
+func (sess *ClientSession) handleConnBroken(connBroken *serverMsg) error {
+	sess.setState(Error)
+	reason := connBroken.Reason
+	err := fmt.Errorf("server broke connection: %s", reason)
+	sess.Log.Errorf("%s", err)
+	switch reason {
+	case protocol.BrokenHostMismatch:
+		sess.resetHosts()
+	}
+	return err
+}
+
 // loop runs the session with the server, emits a stream of events.
 func (sess *ClientSession) loop() error {
 	var err error
@@ -356,6 +374,8 @@
 			err = sess.handlePing()
 		case "broadcast":
 			err = sess.handleBroadcast(&recv)
+		case "connbroken":
+			err = sess.handleConnBroken(&recv)
 		}
 		if err != nil {
 			return err

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-04-03 20:56:25 +0000
+++ client/session/session_test.go	2014-04-11 18:18:05 +0000
@@ -317,6 +317,29 @@
 	c.Check(sess.deliveryHosts, DeepEquals, []string{"baz:443"})
 }
 
+func (cs *clientSessionSuite) TestGetHostsRemoteCachingReset(c *C) {
+	hostGetter := &testHostGetter{[]string{"foo:443", "bar:443"}, nil}
+	sess := &ClientSession{
+		getHost: hostGetter,
+		ClientSessionConfig: ClientSessionConfig{
+			HostsCachingExpiryTime: 2 * time.Hour,
+		},
+		timeSince: time.Since,
+	}
+	err := sess.getHosts()
+	c.Assert(err, IsNil)
+	hostGetter.hosts = []string{"baz:443"}
+	// cached
+	err = sess.getHosts()
+	c.Assert(err, IsNil)
+	c.Check(sess.deliveryHosts, DeepEquals, []string{"foo:443", "bar:443"})
+	// reset
+	sess.resetHosts()
+	err = sess.getHosts()
+	c.Assert(err, IsNil)
+	c.Check(sess.deliveryHosts, DeepEquals, []string{"baz:443"})
+}
+
 /****************************************************************
   startConnectionAttempt()/nextHostToTry()/started tests
 ****************************************************************/
@@ -587,7 +610,7 @@
 				json.RawMessage("false"), // shouldn't happen but robust
 				json.RawMessage(`{"img1/m1":[102,"tubular"]}`),
 			},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
@@ -618,7 +641,7 @@
 			ChanId:   "0",
 			TopLevel: 2,
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	failure := errors.New("ACK ACK ACK")
@@ -635,7 +658,7 @@
 			ChanId:   "something awful",
 			TopLevel: 2,
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
@@ -652,7 +675,7 @@
 			ChanId:   "0",
 			TopLevel: 2,
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	s.upCh <- nil // ack ok
 	// start returns with error
@@ -665,6 +688,37 @@
 }
 
 /****************************************************************
+  handleConnBroken() tests
+****************************************************************/
+
+func (s *msgSuite) TestHandleConnBrokenUnkwown(c *C) {
+	msg := serverMsg{"connbroken",
+		protocol.BroadcastMsg{}, protocol.NotificationsMsg{},
+		protocol.ConnBrokenMsg{
+			Reason: "REASON",
+		},
+	}
+	go func() { s.errCh <- s.sess.handleConnBroken(&msg) }()
+	c.Check(<-s.errCh, ErrorMatches, "server broke connection: REASON")
+	c.Check(s.sess.State(), Equals, Error)
+}
+
+func (s *msgSuite) TestHandleConnBrokenHostMismatch(c *C) {
+	msg := serverMsg{"connbroken",
+		protocol.BroadcastMsg{}, protocol.NotificationsMsg{},
+		protocol.ConnBrokenMsg{
+			Reason: protocol.BrokenHostMismatch,
+		},
+	}
+	s.sess.deliveryHosts = []string{"foo:443", "bar:443"}
+	go func() { s.errCh <- s.sess.handleConnBroken(&msg) }()
+	c.Check(<-s.errCh, ErrorMatches, "server broke connection: host-mismatch")
+	c.Check(s.sess.State(), Equals, Error)
+	// hosts were reset
+	c.Check(s.sess.deliveryHosts, IsNil)
+}
+
+/****************************************************************
   loop() tests
 ****************************************************************/
 
@@ -728,6 +782,17 @@
 	c.Check(<-s.errCh, Equals, failure)
 }
 
+func (s *loopSuite) TestLoopConnBroken(c *C) {
+	c.Check(s.sess.State(), Equals, Running)
+	broken := protocol.ConnBrokenMsg{
+		Type:   "connbroken",
+		Reason: "REASON",
+	}
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	s.upCh <- broken
+	c.Check(<-s.errCh, NotNil)
+}
+
 /****************************************************************
   start() tests
 ****************************************************************/

=== modified file 'debian/changelog'
--- debian/changelog	2014-04-04 14:52:06 +0000
+++ debian/changelog	2014-04-11 18:18:05 +0000
@@ -1,3 +1,4 @@
+<<<<<<< TREE
 ubuntu-push (0.1+14.04.20140404-0ubuntu1) trusty; urgency=low
 
   [ Tarmac ]
@@ -5,6 +6,14 @@
 
  -- Ubuntu daily release <[e-mail]>  Fri, 04 Apr 2014 14:52:06 +0000
 
+=======
+ubuntu-push (0.2-0ubuntu1) UNRELEASED; urgency=medium
+
+  * New upstream release.
+
+ -- John Lenton <[e-mail]>  Fri, 11 Apr 2014 11:19:42 +0100
+
+>>>>>>> MERGE-SOURCE
 ubuntu-push (0.1+14.04.20140327-0ubuntu1) trusty; urgency=medium
 
   [ John Lenton ]

=== modified file 'debian/copyright'
--- debian/copyright	2014-03-24 15:31:42 +0000
+++ debian/copyright	2014-04-11 18:18:05 +0000
@@ -11,7 +11,7 @@
 License: BSD-3-clause
 
 Files: external/murmur3/*
-Copyright: 2013 Sébastien Paolacci
+Copyright: 2013 Sébastien Paolacci
 License: BSD-3-clause
 
 License: GPL-3.0

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-03-19 22:31:20 +0000
+++ protocol/messages.go	2014-04-11 18:18:05 +0000
@@ -49,6 +49,27 @@
 	PingInterval string
 }
 
+// SplittableMsg are messages that may require and are capable of splitting.
+type SplittableMsg interface {
+	Split() (done bool)
+}
+
+// CONNBROKEN message, server side is breaking the connection for reason.
+type ConnBrokenMsg struct {
+	Type string `json:"T"`
+	// reason
+	Reason string
+}
+
+func (m *ConnBrokenMsg) Split() bool {
+	return true
+}
+
+// CONNBROKEN reasons
+const (
+	BrokenHostMismatch = "host-mismatch"
+)
+
 // PING/PONG messages
 type PingPongMsg struct {
 	Type string `json:"T"`
@@ -56,11 +77,6 @@
 
 const maxPayloadSize = 62 * 1024
 
-// SplittableMsg are messages that may require and are capable of splitting.
-type SplittableMsg interface {
-	Split() (done bool)
-}
-
 // BROADCAST messages
 type BroadcastMsg struct {
 	Type      string `json:"T"`

=== modified file 'protocol/messages_test.go'
--- protocol/messages_test.go	2014-02-26 16:04:57 +0000
+++ protocol/messages_test.go	2014-04-11 18:18:05 +0000
@@ -103,3 +103,7 @@
 	b.Reset()
 	c.Check(b.splitting, Equals, 0)
 }
+
+func (s *messagesSuite) TestSplitConnBrokenMsg(c *C) {
+	c.Check((&ConnBrokenMsg{}).Split(), Equals, true)
+}

=== modified file 'sampleconfigs/dev.json'
--- sampleconfigs/dev.json	2014-03-31 16:48:00 +0000
+++ sampleconfigs/dev.json	2014-04-11 18:18:05 +0000
@@ -8,5 +8,6 @@
     "cert_pem_file": "../server/acceptance/ssl/testing.cert",
     "http_addr": "127.0.0.1:8080",
     "http_read_timeout": "5s",
-    "http_write_timeout": "5s"
+    "http_write_timeout": "5s",
+    "delivery_domain": "localhost"
 }

=== modified file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	2014-02-26 19:50:46 +0000
+++ server/acceptance/acceptance_test.go	2014-04-11 18:18:05 +0000
@@ -34,6 +34,7 @@
 	cfg := make(map[string]interface{})
 	suites.FillServerConfig(cfg, addr)
 	suites.FillHTTPServerConfig(cfg, httpAddr)
+	cfg["delivery_domain"] = "localhost"
 	return cfg
 }
 

=== modified file 'server/acceptance/acceptanceclient.go'
--- server/acceptance/acceptanceclient.go	2014-04-03 16:47:47 +0000
+++ server/acceptance/acceptanceclient.go	2014-04-11 18:18:05 +0000
@@ -42,7 +42,8 @@
 	CertPEMBlock    []byte
 	ReportPings     bool
 	Levels          map[string]int64
-	Insecure        bool // don't verify certs
+	Insecure        bool   // don't verify certs
+	Prefix          string // prefix for events
 	// connection
 	Connection net.Conn
 }
@@ -105,7 +106,7 @@
 	if err != nil {
 		return err
 	}
-	events <- fmt.Sprintf("connected %v", conn.LocalAddr())
+	events <- fmt.Sprintf("%sconnected %v", sess.Prefix, conn.LocalAddr())
 	var recv serverMsg
 	for {
 		deadAfter := pingInterval + sess.ExchangeTimeout
@@ -122,7 +123,7 @@
 				return err
 			}
 			if sess.ReportPings {
-				events <- "Ping"
+				events <- sess.Prefix + "ping"
 			}
 		case "broadcast":
 			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
@@ -134,7 +135,7 @@
 			if err != nil {
 				return err
 			}
-			events <- fmt.Sprintf("broadcast chan:%v app:%v topLevel:%d payloads:%s", recv.ChanId, recv.AppId, recv.TopLevel, pack)
+			events <- fmt.Sprintf("%sbroadcast chan:%v app:%v topLevel:%d payloads:%s", sess.Prefix, recv.ChanId, recv.AppId, recv.TopLevel, pack)
 		}
 	}
 	return nil

=== modified file 'server/acceptance/cmd/acceptanceclient.go'
--- server/acceptance/cmd/acceptanceclient.go	2014-04-03 16:47:47 +0000
+++ server/acceptance/cmd/acceptanceclient.go	2014-04-11 18:18:05 +0000
@@ -19,6 +19,7 @@
 
 import (
 	"flag"
+	"fmt"
 	"log"
 	"os"
 	"path/filepath"
@@ -43,6 +44,10 @@
 }
 
 func main() {
+	flag.Usage = func() {
+		fmt.Fprintf(os.Stderr, "Usage: acceptancclient [options] <config.json> <device id>\n")
+		flag.PrintDefaults()
+	}
 	flag.Parse()
 	narg := flag.NArg()
 	switch {
@@ -72,9 +77,11 @@
 		Insecure:     *insecureFlag,
 	}
 	log.Printf("with: %#v", session)
-	session.CertPEMBlock, err = config.LoadFile(cfg.CertPEMFile, filepath.Dir(configFName))
-	if err != nil {
-		log.Fatalf("reading CertPEMFile: %v", err)
+	if !*insecureFlag {
+		session.CertPEMBlock, err = config.LoadFile(cfg.CertPEMFile, filepath.Dir(configFName))
+		if err != nil {
+			log.Fatalf("reading CertPEMFile: %v", err)
+		}
 	}
 	err = session.Dial()
 	if err != nil {

=== modified file 'server/acceptance/suites/broadcast.go'
--- server/acceptance/suites/broadcast.go	2014-04-03 16:47:47 +0000
+++ server/acceptance/suites/broadcast.go	2014-04-11 18:18:05 +0000
@@ -24,6 +24,7 @@
 
 	. "launchpad.net/gocheck"
 
+	"launchpad.net/ubuntu-push/client/gethosts"
 	"launchpad.net/ubuntu-push/protocol"
 	"launchpad.net/ubuntu-push/server/api"
 )
@@ -66,8 +67,6 @@
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	// xxx don't send this one
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[]`)
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":20}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
@@ -261,3 +260,12 @@
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
+
+// test /delivery-hosts
+
+func (s *BroadcastAcceptanceSuite) TestGetHosts(c *C) {
+	gh := gethosts.New("", s.ServerAPIURL+"/delivery-hosts", 2*time.Second)
+	hosts, err := gh.Get()
+	c.Assert(err, IsNil)
+	c.Check(hosts, DeepEquals, []string{s.ServerAddr})
+}

=== modified file 'server/acceptance/suites/pingpong.go'
--- server/acceptance/suites/pingpong.go	2014-04-03 16:47:47 +0000
+++ server/acceptance/suites/pingpong.go	2014-04-11 18:18:05 +0000
@@ -56,11 +56,11 @@
 	c.Assert(connectSrv, Matches, ".*session.* connected .*")
 	c.Assert(registeredSrv, Matches, ".*session.* registered DEVA")
 	c.Assert(strings.HasSuffix(connectSrv, connectCli), Equals, true)
-	c.Assert(NextEvent(events, errCh), Equals, "Ping")
+	c.Assert(NextEvent(events, errCh), Equals, "ping")
 	elapsedOfPing := float64(time.Since(tconnect)) / float64(500*time.Millisecond)
 	c.Check(elapsedOfPing >= 1.0, Equals, true)
 	c.Check(elapsedOfPing < 1.05, Equals, true)
-	c.Assert(NextEvent(events, errCh), Equals, "Ping")
+	c.Assert(NextEvent(events, errCh), Equals, "ping")
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, ".*session.* ended with: EOF")
 	c.Check(len(errCh), Equals, 0)
 }
@@ -87,7 +87,7 @@
 	c.Assert(NextEvent(events, errCh), Matches, "connected .*")
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, ".*session.* connected .*")
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, ".*session.* registered .*")
-	c.Assert(NextEvent(events, errCh), Equals, "Ping")
+	c.Assert(NextEvent(events, errCh), Equals, "ping")
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*timeout`)
 	c.Check(len(errCh), Equals, 0)
 }

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-04-03 14:31:10 +0000
+++ server/broker/broker.go	2014-04-11 18:18:05 +0000
@@ -19,6 +19,7 @@
 package broker
 
 import (
+	"errors"
 	"fmt"
 
 	"launchpad.net/ubuntu-push/protocol"
@@ -46,6 +47,9 @@
 	Acked(sess BrokerSession, done bool) error
 }
 
+// ErrNop returned by Prepare means nothing to do/send.
+var ErrNop = errors.New("nothing to send")
+
 // LevelsMap is the type for holding channel levels for session.
 type LevelsMap map[store.InternalChannelId]int64
 

=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-04-03 16:00:53 +0000
+++ server/broker/exchanges.go	2014-04-11 18:18:05 +0000
@@ -28,8 +28,9 @@
 
 // Scratch area for exchanges, sessions should hold one of these.
 type ExchangesScratchArea struct {
-	broadcastMsg protocol.BroadcastMsg
-	ackMsg       protocol.AckMsg
+	broadcastMsg  protocol.BroadcastMsg
+	ackMsg        protocol.AckMsg
+	connBrokenMsg protocol.ConnBrokenMsg
 }
 
 // BroadcastExchange leads a session through delivering a BROADCAST.
@@ -42,7 +43,7 @@
 }
 
 // check interface already here
-var _ Exchange = &BroadcastExchange{}
+var _ Exchange = (*BroadcastExchange)(nil)
 
 // Init ensures the BroadcastExchange is fully initialized for the sessions.
 func (sbe *BroadcastExchange) Init() {
@@ -88,14 +89,18 @@
 
 // Prepare session for a BROADCAST.
 func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
-	scratchArea := sess.ExchangeScratchArea()
-	scratchArea.broadcastMsg.Reset()
-	scratchArea.broadcastMsg.Type = "broadcast"
 	clientLevel := sess.Levels()[sbe.ChanId]
 	payloads := filterByLevel(clientLevel, sbe.TopLevel, sbe.NotificationPayloads)
 	tag := fmt.Sprintf("%s/%s", sess.DeviceImageChannel(), sess.DeviceImageModel())
 	payloads = channelFilter(tag, sbe.ChanId, payloads, sbe.Decoded)
+	if len(payloads) == 0 && sbe.TopLevel >= clientLevel {
+		// empty and don't need to force resync => do nothing
+		return nil, nil, ErrNop
+	}
 
+	scratchArea := sess.ExchangeScratchArea()
+	scratchArea.broadcastMsg.Reset()
+	scratchArea.broadcastMsg.Type = "broadcast"
 	// xxx need an AppId as well, later
 	scratchArea.broadcastMsg.ChanId = store.InternalChannelIdToHex(sbe.ChanId)
 	scratchArea.broadcastMsg.TopLevel = sbe.TopLevel
@@ -113,3 +118,24 @@
 	sess.Levels()[sbe.ChanId] = sbe.TopLevel
 	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")
+}

=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-04-04 09:57:02 +0000
+++ server/broker/exchanges_test.go	2014-04-11 18:18:05 +0000
@@ -81,6 +81,44 @@
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(3))
 }
 
+func (s *exchangesSuite) TestBroadcastExchangeEmpty(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		Model:        "m1",
+		ImageChannel: "img1",
+	}
+	exchg := &broker.BroadcastExchange{
+		ChanId:               store.SystemInternalChannelId,
+		TopLevel:             3,
+		NotificationPayloads: []json.RawMessage{},
+	}
+	exchg.Init()
+	outMsg, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, Equals, broker.ErrNop)
+	c.Check(outMsg, IsNil)
+	c.Check(inMsg, IsNil)
+}
+
+func (s *exchangesSuite) TestBroadcastExchangeEmptyButAhead(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 10,
+		}),
+		Model:        "m1",
+		ImageChannel: "img1",
+	}
+	exchg := &broker.BroadcastExchange{
+		ChanId:               store.SystemInternalChannelId,
+		TopLevel:             3,
+		NotificationPayloads: []json.RawMessage{},
+	}
+	exchg.Init()
+	outMsg, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	c.Check(outMsg, NotNil)
+	c.Check(inMsg, NotNil)
+}
+
 func (s *exchangesSuite) TestBroadcastExchangeReuseVsSplit(c *C) {
 	sess := &testing.TestBrokerSession{
 		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
@@ -210,3 +248,17 @@
 	c.Assert(err, IsNil)
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(5))
 }
+
+func (s *exchangesSuite) TestConnBrokenExchange(c *C) {
+	sess := &testing.TestBrokerSession{}
+	cbe := &broker.ConnBrokenExchange{"REASON"}
+	outMsg, inMsg, err := cbe.Prepare(sess)
+	c.Assert(err, IsNil)
+	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")
+}

=== modified file 'server/broker/simple/simple.go'
--- server/broker/simple/simple.go	2014-04-03 16:00:53 +0000
+++ server/broker/simple/simple.go	2014-04-11 18:18:05 +0000
@@ -222,6 +222,10 @@
 					delete(b.registry, sess.deviceId)
 				}
 			} else { // register
+				prev := b.registry[sess.deviceId]
+				if prev != nil { // kick it
+					close(prev.exchanges)
+				}
 				b.registry[sess.deviceId] = sess
 				sess.registered = true
 				sess.done <- true

=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-04-03 16:00:53 +0000
+++ server/broker/testsuite/suite.go	2014-04-11 18:18:05 +0000
@@ -164,6 +164,14 @@
 	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
+	select {
+	case _, ok := <-sess1.SessionChannel():
+		checkAndFalse = ok == false
+	default:
+	}
+	c.Check(checkAndFalse, Equals, true)
 	c.Assert(s.RevealSession(b, "dev-1"), Equals, sess2)
 	b.Unregister(sess1)
 	// just to make sure the unregister was processed

=== modified file 'server/dev/server.go'
--- server/dev/server.go	2014-03-25 19:02:18 +0000
+++ server/dev/server.go	2014-04-11 18:18:05 +0000
@@ -18,6 +18,7 @@
 package main
 
 import (
+	"encoding/json"
 	"net"
 	"net/http"
 	"os"
@@ -37,6 +38,8 @@
 	server.DevicesParsedConfig
 	// api http server configuration
 	server.HTTPServeParsedConfig
+	// delivery domain
+	DeliveryDomain string `json:"delivery_domain"`
 }
 
 func main() {
@@ -60,11 +63,25 @@
 	storeForRequest := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
 		return sto, nil
 	}
+	lst, err := net.Listen("tcp", cfg.Addr())
+	if err != nil {
+		server.BootLogFatalf("start device listening: %v", err)
+	}
 	mux := api.MakeHandlersMux(storeForRequest, broker, logger)
+	// & /delivery-hosts
+	mux.HandleFunc("/delivery-hosts", func(w http.ResponseWriter, req *http.Request) {
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("Content-Type", "application/json")
+		enc := json.NewEncoder(w)
+		enc.Encode(map[string]interface{}{
+			"hosts":  []string{lst.Addr().String()},
+			"domain": cfg.DeliveryDomain,
+		})
+	})
 	handler := api.PanicTo500Handler(mux, logger)
 	go server.HTTPServeRunner(nil, handler, &cfg.HTTPServeParsedConfig)()
 	// listen for device connections
-	server.DevicesRunner(nil, func(conn net.Conn) error {
+	server.DevicesRunner(lst, func(conn net.Conn) error {
 		track := session.NewTracker(logger)
 		return session.Session(conn, broker, cfg, track)
 	}, logger, &cfg.DevicesParsedConfig)()

=== modified file 'server/session/session.go'
--- server/session/session.go	2014-02-10 23:19:08 +0000
+++ server/session/session.go	2014-04-11 18:18:05 +0000
@@ -62,6 +62,9 @@
 	if err != nil {
 		return err
 	}
+	if inMsg == nil { // no answer expected, breaking connection
+		return &broker.ErrAbort{"session broken for reason"}
+	}
 	err = proto.ReadMessage(inMsg)
 	if err != nil {
 		return err
@@ -76,6 +79,7 @@
 	pingTimer := time.NewTimer(pingInterval)
 	intervalStart := time.Now()
 	ch := sess.SessionChannel()
+Loop:
 	for {
 		select {
 		case <-pingTimer.C:
@@ -90,10 +94,17 @@
 				return &broker.ErrAbort{"expected PONG message"}
 			}
 			pingTimer.Reset(pingInterval)
-		case exchg := <-ch:
-			// xxx later can use ch closing for shutdown/reset
+		case exchg, ok := <-ch:
 			pingTimer.Stop()
+			if !ok {
+				return &broker.ErrAbort{"terminated"}
+			}
 			outMsg, inMsg, err := exchg.Prepare(sess)
+			if err == broker.ErrNop { // nothing to do
+				pingTimer.Reset(pingInterval)
+				intervalStart = time.Now()
+				continue Loop
+			}
 			if err != nil {
 				return err
 			}

=== modified file 'server/session/session_test.go'
--- server/session/session_test.go	2014-03-19 23:46:18 +0000
+++ server/session/session_test.go	2014-04-11 18:18:05 +0000
@@ -346,6 +346,42 @@
 	c.Check(err, Equals, io.EOF)
 }
 
+func (s *sessionSuite) TestSessionLoopKick(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)
+	sess := &testing.TestBrokerSession{Exchanges: exchanges}
+	go func() {
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
+	}()
+	close(exchanges)
+	err := <-errCh
+	c.Check(err, DeepEquals, &broker.ErrAbort{"terminated"})
+}
+
+func (s *sessionSuite) TestSessionLoopExchangeErrNop(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)
+	exchanges <- &testExchange{prepErr: broker.ErrNop}
+	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.PingPongMsg{Type: "ping"})
+	up <- nil // no write error
+	up <- io.EOF
+	err := <-errCh
+	c.Check(err, Equals, io.EOF)
+}
+
 func (s *sessionSuite) TestSessionLoopExchangeSplit(c *C) {
 	nopTrack := NewTracker(s.testlog)
 	errCh := make(chan error, 1)
@@ -434,6 +470,25 @@
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
 }
 
+func (s *sessionSuite) TestSessionLoopConnBrokenExchange(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)
+	exchanges <- &broker.ConnBrokenExchange{"REASON"}
+	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"})
+	up <- nil // no write error
+	err := <-errCh
+	c.Check(err, DeepEquals, &broker.ErrAbort{"session broken for reason"})
+}
+
 type testTracker struct {
 	SessionTracker
 	interval chan interface{}