Merge lp:~pedronis/ubuntu-push/fix-split-reuse into lp:ubuntu-push

[Samuele Pedroni]

* fix the fact that exchanges were reusing broadcast messages but the split state of those wasn't reset
* fix naming in tests that was reversed

* fix the fact that exchanges were reusing broadcast messages but the split state of those wasn't reset

* fix naming in tests that was reversed

[Nicola Larosa (teknico)]

LGTM

review: Approve

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-02-24 15:54:37 +0000
+++ protocol/messages.go	2014-02-26 16:06:13 +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 16:06:13 +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/broker/exchanges.go'
--- server/broker/exchanges.go	2014-02-10 23:19:08 +0000
+++ server/broker/exchanges.go	2014-02-26 16:06:13 +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 16:06:13 +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)