Merge lp:~pedronis/ubuntu-push/handle-levels-that-are-off into lp:ubuntu-push

[Samuele Pedroni]

fix corner cases like client level too ahead etc, with acceptance tests

fix corner cases like client level too ahead etc, with acceptance tests

[lp:~pedronis/ubuntu-push/handle-levels-that-are-off]

46. By Samuele Pedroni on 2014-02-04

fix corner cases, client too ahead etc, with acceptance tests

[John Lenton (chipaca)]

Thank you!

review: Approve

=== modified file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	2014-01-30 08:56:57 +0000
+++ server/acceptance/acceptance_test.go	2014-02-04 12:25:11 +0000
@@ -484,3 +484,92 @@
 	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
 	c.Check(len(errCh), Equals, 0)
 }
+
+func (s *acceptanceSuite) TestBroadcastTooAhead(c *C) {
+	// send broadcasts that will be pending
+	got, err := s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 1}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+	got, err = s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 2}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+
+	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
+	}
+	events, errCh := s.startClient(c, "DEVB", intercept, 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}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestBroadcastTooAheadOnEmpty(c *C) {
+	// nothing there
+	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
+	}
+	events, errCh := s.startClient(c, "DEVB", intercept, 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`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestBroadcastWayBehind(c *C) {
+	// send broadcasts that will be pending
+	got, err := s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 1}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+	got, err = s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 2}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+
+	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
+	}
+	events, errCh := s.startClient(c, "DEVB", intercept, 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}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}

=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-01-31 16:36:16 +0000
+++ server/broker/exchanges.go	2014-02-04 12:25:11 +0000
@@ -44,7 +44,13 @@
 
 func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
 	c := int64(len(payloads))
+	if c == 0 {
+		return nil
+	}
 	delta := topLevel - clientLevel
+	if delta < 0 { // means too ahead, send the last pending
+		delta = 1
+	}
 	if delta < c {
 		return payloads[c-delta:]
 	} else {

=== modified file 'server/broker/exchg_impl_test.go'
--- server/broker/exchg_impl_test.go	2014-01-23 20:13:22 +0000
+++ server/broker/exchg_impl_test.go	2014-02-04 12:25:11 +0000
@@ -44,4 +44,15 @@
 	c.Check(len(res), Equals, 3)
 	res = filterByLevel(1, 5, payloads)
 	c.Check(len(res), Equals, 3)
+	// too ahead, pick only last
+	res = filterByLevel(10, 5, payloads)
+	c.Check(len(res), Equals, 1)
+	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 5}`))
+}
+
+func (s *exchangesImplSuite) TestFilterByLevelEmpty(c *C) {
+	res := filterByLevel(5, 0, nil)
+	c.Check(len(res), Equals, 0)
+	res = filterByLevel(5, 10, nil)
+	c.Check(len(res), Equals, 0)
 }