Merge lp:~pedronis/ubuntu-push/expose-more-acceptance-details into lp:ubuntu-push

[Samuele Pedroni]

expose more bits of acceptance tests

[lp:~pedronis/ubuntu-push/expose-more-acceptance-details]

78. By Samuele Pedroni on 2014-02-26

Merged fix-split-reuse into expose-more-acceptance-details.

79. By Samuele Pedroni on 2014-02-26

log panics from servers in full in tests

80. By Samuele Pedroni on 2014-02-26

change the signature of StartServer

Unmerged revisions

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-02-24 15:54:37 +0000
+++ protocol/messages.go	2014-02-26 19:50:48 +0000
@@ -93,6 +93,12 @@
 	return true
 }
 
+// Reset resets the splitting state if the message storage is to be
+// reused.
+func (b *BroadcastMsg) Reset() {
+	b.splitting = 0
+}
+
 // NOTIFICATIONS message
 type NotificationsMsg struct {
 	Type          string `json:"T"`
 
=== modified file 'protocol/messages_test.go'
--- protocol/messages_test.go	2014-02-10 22:53:00 +0000
+++ protocol/messages_test.go	2014-02-26 19:50:48 +0000
@@ -99,4 +99,7 @@
 	n3 := len(b.Payloads)
 	c.Check(b.TopLevel, Equals, int64(n))
 	c.Check(n1+n2+n3, Equals, n)
+	// reset
+	b.Reset()
+	c.Check(b.splitting, Equals, 0)
 }
 
=== modified file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	2014-02-13 19:33:14 +0000
+++ server/acceptance/acceptance_test.go	2014-02-26 19:50:48 +0000
@@ -38,7 +38,7 @@
 }
 
 // Start a server.
-func StartServer(c *C) (<-chan string, func(), string, string) {
+func StartServer(c *C, s *suites.AcceptanceSuite, handle *suites.ServerHandle) {
 	if *serverCmd == "" {
 		c.Skip("executable server not specified")
 	}
@@ -46,10 +46,11 @@
 	cfg := testServerConfig("127.0.0.1:0", "127.0.0.1:0")
 	cfgFilename := suites.WriteConfig(c, tmpDir, "config.json", cfg)
 	logs, killServer := suites.RunAndObserve(c, *serverCmd, cfgFilename)
-	serverHTTPAddr := suites.ExtractListeningAddr(c, logs, suites.HTTPListeningOnPat)
-	serverURL := fmt.Sprintf("http://%s", serverHTTPAddr)
-	serverAddr := suites.ExtractListeningAddr(c, logs, suites.DevListeningOnPat)
-	return logs, killServer, serverAddr, serverURL
+	s.KillGroup["server"] = killServer
+	handle.ServerHTTPAddr = suites.ExtractListeningAddr(c, logs, suites.HTTPListeningOnPat)
+	s.ServerAPIURL = fmt.Sprintf("http://%s", handle.ServerHTTPAddr)
+	handle.ServerAddr = suites.ExtractListeningAddr(c, logs, suites.DevListeningOnPat)
+	handle.ServerEvents = logs
 }
 
 // ping pong/connectivity
 
=== modified file 'server/acceptance/suites/broadcast.go'
--- server/acceptance/suites/broadcast.go	2014-02-12 15:43:24 +0000
+++ server/acceptance/suites/broadcast.go	2014-02-26 19:50:48 +0000
@@ -37,8 +37,8 @@
 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{
+	events, errCh, stop := s.StartClient(c, "DEVB", nil)
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"n": 42}`),
@@ -47,13 +47,13 @@
 	c.Assert(got, Matches, ".*ok.*")
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastPending(c *C) {
 	// send broadcast that will be pending
-	got, err := s.postRequest("/broadcast", &api.Broadcast{
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 1}`),
@@ -61,11 +61,11 @@
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 
-	events, errCh, stop := s.startClient(c, "DEVB", nil)
+	events, errCh, stop := s.StartClient(c, "DEVB", nil)
 	// gettting pending on connect
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"b":1}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
@@ -73,7 +73,7 @@
 	// send bunch of broadcasts that will be pending
 	payloadFmt := fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
 	for i := 0; i < 32; i++ {
-		got, err := s.postRequest("/broadcast", &api.Broadcast{
+		got, err := s.PostRequest("/broadcast", &api.Broadcast{
 			Channel:  "system",
 			ExpireOn: future,
 			Data:     json.RawMessage(fmt.Sprintf(payloadFmt, i)),
@@ -82,22 +82,22 @@
 		c.Assert(got, Matches, ".*ok.*")
 	}
 
-	events, errCh, stop := s.startClient(c, "DEVC", nil)
+	events, errCh, stop := s.StartClient(c, "DEVC", nil)
 	// gettting pending on connect
 	c.Check(NextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:30 payloads:\[{"b":0,.*`)
 	c.Check(NextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:32 payloads:\[.*`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastDistribution2(c *C) {
-	// start 1st clinet
-	events1, errCh1, stop1 := s.startClient(c, "DEV1", nil)
+	// start 1st client
+	events1, errCh1, stop1 := s.StartClient(c, "DEV1", nil)
 	// start 2nd client
-	events2, errCh2, stop2 := s.startClient(c, "DEV2", nil)
+	events2, errCh2, stop2 := s.StartClient(c, "DEV2", nil)
 	// broadcast
-	got, err := s.postRequest("/broadcast", &api.Broadcast{
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"n": 42}`),
@@ -108,15 +108,15 @@
 	c.Check(NextEvent(events2, errCh2), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
 	stop1()
 	stop2()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh1), Equals, 0)
 	c.Check(len(errCh2), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastFilterByLevel(c *C) {
-	events, errCh, stop := s.startClient(c, "DEVD", nil)
-	got, err := s.postRequest("/broadcast", &api.Broadcast{
+	events, errCh, stop := s.StartClient(c, "DEVD", nil)
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 1}`),
@@ -125,10 +125,10 @@
 	c.Assert(got, Matches, ".*ok.*")
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"b":1}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 	// another broadcast
-	got, err = s.postRequest("/broadcast", &api.Broadcast{
+	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 2}`),
@@ -136,25 +136,25 @@
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 	// reconnect, provide levels, get only later notification
-	events, errCh, stop = s.startClient(c, "DEVD", map[string]int64{
+	events, errCh, stop = s.StartClient(c, "DEVD", map[string]int64{
 		protocol.SystemChannelId: 1,
 	})
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":2}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastTooAhead(c *C) {
 	// send broadcasts that will be pending
-	got, err := s.postRequest("/broadcast", &api.Broadcast{
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	got, err = s.postRequest("/broadcast", &api.Broadcast{
+	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 2}`),
@@ -162,38 +162,38 @@
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 
-	events, errCh, stop := s.startClient(c, "DEVB", map[string]int64{
+	events, errCh, stop := s.StartClient(c, "DEVB", map[string]int64{
 		protocol.SystemChannelId: 10,
 	})
 	// gettting last one pending on connect
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":2}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastTooAheadOnEmpty(c *C) {
 	// nothing there
-	events, errCh, stop := s.startClient(c, "DEVB", map[string]int64{
+	events, errCh, stop := s.StartClient(c, "DEVB", map[string]int64{
 		protocol.SystemChannelId: 10,
 	})
 	// gettting empty pending on connect
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:0 payloads:null`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastWayBehind(c *C) {
 	// send broadcasts that will be pending
-	got, err := s.postRequest("/broadcast", &api.Broadcast{
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	got, err = s.postRequest("/broadcast", &api.Broadcast{
+	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 2}`),
@@ -201,26 +201,26 @@
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
 
-	events, errCh, stop := s.startClient(c, "DEVB", map[string]int64{
+	events, errCh, stop := s.StartClient(c, "DEVB", map[string]int64{
 		protocol.SystemChannelId: -10,
 	})
 	// gettting pending on connect
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":1},{"b":2}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *BroadcastAcceptanceSuite) TestBroadcastExpiration(c *C) {
 	// send broadcast that will be pending, and one that will expire
-	got, err := s.postRequest("/broadcast", &api.Broadcast{
+	got, err := s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"b": 1}`),
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	got, err = s.postRequest("/broadcast", &api.Broadcast{
+	got, err = s.PostRequest("/broadcast", &api.Broadcast{
 		Channel:  "system",
 		ExpireOn: time.Now().Add(1 * time.Second).Format(time.RFC3339),
 		Data:     json.RawMessage(`{"b": 2}`),
@@ -231,10 +231,10 @@
 	time.Sleep(2 * time.Second)
 	// second broadcast is expired
 
-	events, errCh, stop := s.startClient(c, "DEVB", nil)
+	events, errCh, stop := s.StartClient(c, "DEVB", nil)
 	// gettting pending on connect
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":1}]`)
 	stop()
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
 
=== modified file 'server/acceptance/suites/helpers.go'
--- server/acceptance/suites/helpers.go	2014-02-11 20:36:32 +0000
+++ server/acceptance/suites/helpers.go	2014-02-26 19:50:48 +0000
@@ -92,12 +92,15 @@
 		c.Fatal(err)
 	}
 	bufErr := bufio.NewReaderSize(stderr, 5000)
-	getLineInfo := func() (string, error) {
+	getLineInfo := func(full bool) (string, error) {
 		for {
 			line, err := bufErr.ReadString('\n')
 			if err != nil {
 				return "", err
 			}
+			if full {
+				return strings.TrimRight(line, "\n"), nil
+			}
 			extracted := rxLineInfo.FindStringSubmatch(line)
 			if extracted == nil {
 				return "", fmt.Errorf("unexpected line: %#v", line)
@@ -108,13 +111,19 @@
 	}
 	logs := make(chan string, 10)
 	go func() {
+		paniced := false
 		for {
-			info, err := getLineInfo()
+			info, err := getLineInfo(paniced)
 			if err != nil {
 				logs <- fmt.Sprintf("%s capture: %v", cmdName, err)
 				close(logs)
 				return
 			}
+			if paniced || strings.HasPrefix(info, "ERROR(PANIC") {
+				paniced = true
+				c.Log(info)
+				continue
+			}
 			logs <- info
 		}
 	}()
 
=== modified file 'server/acceptance/suites/pingpong.go'
--- server/acceptance/suites/pingpong.go	2014-02-11 20:36:32 +0000
+++ server/acceptance/suites/pingpong.go	2014-02-26 19:50:48 +0000
@@ -34,7 +34,7 @@
 func (s *PingPongAcceptanceSuite) TestConnectPingPing(c *C) {
 	errCh := make(chan error, 1)
 	events := make(chan string, 10)
-	sess := testClientSession(s.serverAddr, "DEVA", true)
+	sess := testClientSession(s.ServerAddr, "DEVA", true)
 	err := sess.Dial()
 	c.Assert(err, IsNil)
 	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
@@ -50,8 +50,8 @@
 		errCh <- sess.Run(events)
 	}()
 	connectCli := NextEvent(events, errCh)
-	connectSrv := NextEvent(s.serverEvents, nil)
-	registeredSrv := NextEvent(s.serverEvents, nil)
+	connectSrv := NextEvent(s.ServerEvents, nil)
+	registeredSrv := NextEvent(s.ServerEvents, nil)
 	tconnect := time.Now()
 	c.Assert(connectSrv, Matches, ".*session.* connected .*")
 	c.Assert(registeredSrv, Matches, ".*session.* registered DEVA")
@@ -61,14 +61,14 @@
 	c.Check(elapsedOfPing >= 1.0, Equals, true)
 	c.Check(elapsedOfPing < 1.05, Equals, true)
 	c.Assert(NextEvent(events, errCh), Equals, "Ping")
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, ".*session.* ended with: EOF")
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, ".*session.* ended with: EOF")
 	c.Check(len(errCh), Equals, 0)
 }
 
 func (s *PingPongAcceptanceSuite) TestConnectPingNeverPong(c *C) {
 	errCh := make(chan error, 1)
 	events := make(chan string, 10)
-	sess := testClientSession(s.serverAddr, "DEVB", true)
+	sess := testClientSession(s.ServerAddr, "DEVB", true)
 	err := sess.Dial()
 	c.Assert(err, IsNil)
 	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
@@ -85,9 +85,9 @@
 		errCh <- sess.Run(events)
 	}()
 	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(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(s.serverEvents, nil), Matches, `.* ended with:.*timeout`)
+	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*timeout`)
 	c.Check(len(errCh), Equals, 0)
 }
 
=== modified file 'server/acceptance/suites/suite.go'
--- server/acceptance/suites/suite.go	2014-02-11 20:36:32 +0000
+++ server/acceptance/suites/suite.go	2014-02-26 19:50:48 +0000
@@ -34,43 +34,79 @@
 	helpers "launchpad.net/ubuntu-push/testing"
 )
 
+// ServerHandle holds the information to attach a client to the test server.
+type ServerHandle struct {
+	ServerAddr     string
+	ServerHTTPAddr string
+	ServerEvents   <-chan string
+}
+
+// Start a client.
+func (h *ServerHandle) StartClient(c *C, devId string, levels map[string]int64) (events <-chan string, errorCh <-chan error, stop func()) {
+	errCh := make(chan error, 1)
+	cliEvents := make(chan string, 10)
+	sess := testClientSession(h.ServerAddr, devId, false)
+	sess.Levels = levels
+	err := sess.Dial()
+	c.Assert(err, IsNil)
+	clientShutdown := make(chan bool, 1) // abused as an atomic flag
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// read after ack
+		if op == "read" && len(clientShutdown) > 0 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	sess.Connection = &interceptingConn{sess.Connection, 0, 0, intercept}
+	go func() {
+		errCh <- sess.Run(cliEvents)
+	}()
+	c.Assert(NextEvent(cliEvents, errCh), Matches, "connected .*")
+	c.Assert(NextEvent(h.ServerEvents, nil), Matches, ".*session.* connected .*")
+	c.Assert(NextEvent(h.ServerEvents, nil), Matches, ".*session.* registered "+devId)
+	return cliEvents, errCh, func() { clientShutdown <- true }
+}
+
 // AcceptanceSuite has the basic functionality of the acceptance suites.
 type AcceptanceSuite struct {
 	// hook to start the server(s)
-	StartServer func(c *C) (logs <-chan string, kill func(), serverAddr, apiURL string)
-	// running bits
-	serverKill   func()
-	serverAddr   string
-	serverAPIURL string
-	serverEvents <-chan string
-	httpClient   *http.Client
+	StartServer func(c *C, s *AcceptanceSuite, handle *ServerHandle)
+	// populated by StartServer
+	ServerHandle
+	ServerAPIURL string
+	// KillGroup should be populated by StartServer with functions
+	// to kill the server process
+	KillGroup map[string]func()
+	// other state
+	httpClient *http.Client
 }
 
 // Start a new server for each test.
 func (s *AcceptanceSuite) SetUpTest(c *C) {
-	logs, kill, addr, url := s.StartServer(c)
-	s.serverEvents = logs
-	s.serverKill = kill
-	s.serverAddr = addr
-	s.serverAPIURL = url
+	s.KillGroup = make(map[string]func())
+	s.StartServer(c, s, &s.ServerHandle)
+	c.Assert(s.ServerHandle.ServerEvents, NotNil)
+	c.Assert(s.ServerHandle.ServerAddr, Not(Equals), "")
+	c.Assert(s.ServerAPIURL, Not(Equals), "")
 	s.httpClient = &http.Client{}
 }
 
 func (s *AcceptanceSuite) TearDownTest(c *C) {
-	if s.serverKill != nil {
-		s.serverKill()
+	for _, f := range s.KillGroup {
+		f()
 	}
 }
 
-// Post a request.
-func (s *AcceptanceSuite) postRequest(path string, message interface{}) (string, error) {
+// Post a API request.
+func (s *AcceptanceSuite) PostRequest(path string, message interface{}) (string, error) {
 	packedMessage, err := json.Marshal(message)
 	if err != nil {
 		panic(err)
 	}
 	reader := bytes.NewReader(packedMessage)
 
-	url := s.serverAPIURL + path
+	url := s.ServerAPIURL + path
 	request, _ := http.NewRequest("POST", url, reader)
 	request.ContentLength = int64(reader.Len())
 	request.Header.Set("Content-Type", "application/json")
@@ -141,30 +177,3 @@
 	}
 	return
 }
-
-// Start a client.
-func (s *AcceptanceSuite) startClient(c *C, devId string, levels map[string]int64) (events <-chan string, errorCh <-chan error, stop func()) {
-	errCh := make(chan error, 1)
-	cliEvents := make(chan string, 10)
-	sess := testClientSession(s.serverAddr, devId, false)
-	sess.Levels = levels
-	err := sess.Dial()
-	c.Assert(err, IsNil)
-	clientShutdown := make(chan bool, 1) // abused as an atomic flag
-	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
-		// read after ack
-		if op == "read" && len(clientShutdown) > 0 {
-			// exit the sess.Run() goroutine, client will close
-			runtime.Goexit()
-		}
-		return false, 0, nil
-	}
-	sess.Connection = &interceptingConn{sess.Connection, 0, 0, intercept}
-	go func() {
-		errCh <- sess.Run(cliEvents)
-	}()
-	c.Assert(NextEvent(cliEvents, errCh), Matches, "connected .*")
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, ".*session.* connected .*")
-	c.Assert(NextEvent(s.serverEvents, nil), Matches, ".*session.* registered "+devId)
-	return cliEvents, errCh, func() { clientShutdown <- true }
-}
 
=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-02-10 23:19:08 +0000
+++ server/broker/exchanges.go	2014-02-26 19:50:48 +0000
@@ -61,6 +61,7 @@
 // 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)
 
=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-02-10 23:19:08 +0000
+++ server/broker/exchanges_test.go	2014-02-26 19:50:48 +0000
@@ -18,6 +18,8 @@
 
 import (
 	"encoding/json"
+	"fmt"
+	"strings"
 	stdtesting "testing"
 
 	. "launchpad.net/gocheck"
@@ -45,19 +47,60 @@
 			json.RawMessage(`{"a":"y"}`),
 		},
 	}
-	inMsg, outMsg, err := exchg.Prepare(sess)
+	outMsg, inMsg, err := exchg.Prepare(sess)
 	c.Assert(err, IsNil)
 	// check
-	marshalled, err := json.Marshal(inMsg)
+	marshalled, err := json.Marshal(outMsg)
 	c.Assert(err, IsNil)
 	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"x"},{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
 	c.Assert(err, IsNil)
 	err = exchg.Acked(sess, true)
 	c.Assert(err, IsNil)
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(3))
 }
 
+func (s *exchangesSuite) TestBroadcastExchangeReuseVsSplit(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{}),
+	}
+	payloadFmt := fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
+	needsSplitting := make([]json.RawMessage, 32)
+	for i := 0; i < 32; i++ {
+		needsSplitting[i] = json.RawMessage(fmt.Sprintf(payloadFmt, i))
+	}
+
+	topLevel := int64(len(needsSplitting))
+	exchg := &broker.BroadcastExchange{
+		ChanId:               store.SystemInternalChannelId,
+		TopLevel:             topLevel,
+		NotificationPayloads: needsSplitting,
+	}
+	outMsg, _, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	parts := 0
+	for {
+		done := outMsg.Split()
+		parts++
+		if done {
+			break
+		}
+	}
+	c.Assert(parts, Equals, 2)
+	exchg = &broker.BroadcastExchange{
+		ChanId:   store.SystemInternalChannelId,
+		TopLevel: topLevel + 2,
+		NotificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"x"}`),
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	outMsg, _, err = exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	done := outMsg.Split() // shouldn't panic
+	c.Check(done, Equals, true)
+}
+
 func (s *exchangesSuite) TestBroadcastExchangeAckMismatch(c *C) {
 	sess := &testing.TestBrokerSession{
 		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{}),
@@ -69,13 +112,13 @@
 			json.RawMessage(`{"a":"y"}`),
 		},
 	}
-	inMsg, outMsg, err := exchg.Prepare(sess)
+	outMsg, inMsg, err := exchg.Prepare(sess)
 	c.Assert(err, IsNil)
 	// check
-	marshalled, err := json.Marshal(inMsg)
+	marshalled, err := json.Marshal(outMsg)
 	c.Assert(err, IsNil)
 	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{}`), outMsg)
+	err = json.Unmarshal([]byte(`{}`), inMsg)
 	c.Assert(err, IsNil)
 	err = exchg.Acked(sess, true)
 	c.Assert(err, Not(IsNil))
@@ -96,13 +139,13 @@
 			json.RawMessage(`{"a":"y"}`),
 		},
 	}
-	inMsg, outMsg, err := exchg.Prepare(sess)
+	outMsg, inMsg, err := exchg.Prepare(sess)
 	c.Assert(err, IsNil)
 	// check
-	marshalled, err := json.Marshal(inMsg)
+	marshalled, err := json.Marshal(outMsg)
 	c.Assert(err, IsNil)
 	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
-	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), inMsg)
 	c.Assert(err, IsNil)
 	err = exchg.Acked(sess, true)
 	c.Assert(err, IsNil)