Merge pull request #2048 from calmh/clusterconfigrace

Make sure connection is added to m.protoConn and m.rawConn before it's Start()ed (fixes #2034)

Godeps/Godeps.json

@@ -35,7 +35,7 @@
 		},
 		{
 			"ImportPath": "github.com/syncthing/protocol",
-			"Rev": "9dd6f848bdcd3550606158a33d6aa98de6ea0cdc"
+			"Rev": "b29cfce29e9af56b07b311d27eedceff805f92e2"
 		},
 		{
 			"ImportPath": "github.com/syndtr/goleveldb/leveldb",

Godeps/_workspace/src/github.com/syncthing/protocol/protocol.go

@@ -89,6 +89,7 @@ type Model interface {
 }
 
 type Connection interface {
+	Start()
 	ID() DeviceID
 	Name() string
 	Index(folder string, files []FileInfo, flags uint32, options []Option) error
@@ -161,12 +162,16 @@ func NewConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, receiv
 		compression: compress,
 	}
 
+	return wireFormatConnection{&c}
+}
+
+// Start creates the goroutines for sending a receiving of messages. It must
+// be called exactly once after creating a connection.
+func (c *rawConnection) Start() {
 	go c.readerLoop()
 	go c.writerLoop()
 	go c.pingerLoop()
 	go c.idGenerator()
-
-	return wireFormatConnection{&c}
 }
 
 func (c *rawConnection) ID() DeviceID {

Godeps/_workspace/src/github.com/syncthing/protocol/protocol_test.go

@@ -68,7 +68,9 @@ func TestPing(t *testing.T) {
 	br, bw := io.Pipe()
 
 	c0 := NewConnection(c0ID, ar, bw, newTestModel(), "name", CompressAlways).(wireFormatConnection).next.(*rawConnection)
+	c0.Start()
 	c1 := NewConnection(c1ID, br, aw, newTestModel(), "name", CompressAlways).(wireFormatConnection).next.(*rawConnection)
+	c1.Start()
 	c0.ClusterConfig(ClusterConfigMessage{})
 	c1.ClusterConfig(ClusterConfigMessage{})
 
@@ -94,7 +96,9 @@ func TestPingErr(t *testing.T) {
 			ebw := &ErrPipe{PipeWriter: *bw, max: j, err: e}
 
 			c0 := NewConnection(c0ID, ar, ebw, m0, "name", CompressAlways).(wireFormatConnection).next.(*rawConnection)
+			c0.Start()
 			c1 := NewConnection(c1ID, br, eaw, m1, "name", CompressAlways)
+			c1.Start()
 			c0.ClusterConfig(ClusterConfigMessage{})
 			c1.ClusterConfig(ClusterConfigMessage{})
 
@@ -174,7 +178,9 @@ func TestVersionErr(t *testing.T) {
 	br, bw := io.Pipe()
 
 	c0 := NewConnection(c0ID, ar, bw, m0, "name", CompressAlways).(wireFormatConnection).next.(*rawConnection)
+	c0.Start()
 	c1 := NewConnection(c1ID, br, aw, m1, "name", CompressAlways)
+	c1.Start()
 	c0.ClusterConfig(ClusterConfigMessage{})
 	c1.ClusterConfig(ClusterConfigMessage{})
 
@@ -199,7 +205,9 @@ func TestTypeErr(t *testing.T) {
 	br, bw := io.Pipe()
 
 	c0 := NewConnection(c0ID, ar, bw, m0, "name", CompressAlways).(wireFormatConnection).next.(*rawConnection)
+	c0.Start()
 	c1 := NewConnection(c1ID, br, aw, m1, "name", CompressAlways)
+	c1.Start()
 	c0.ClusterConfig(ClusterConfigMessage{})
 	c1.ClusterConfig(ClusterConfigMessage{})
 
@@ -224,7 +232,9 @@ func TestClose(t *testing.T) {
 	br, bw := io.Pipe()
 
 	c0 := NewConnection(c0ID, ar, bw, m0, "name", CompressAlways).(wireFormatConnection).next.(*rawConnection)
+	c0.Start()
 	c1 := NewConnection(c1ID, br, aw, m1, "name", CompressAlways)
+	c1.Start()
 	c0.ClusterConfig(ClusterConfigMessage{})
 	c1.ClusterConfig(ClusterConfigMessage{})
 

Godeps/_workspace/src/github.com/syncthing/protocol/wireformat.go

@@ -12,6 +12,10 @@ type wireFormatConnection struct {
 	next Connection
 }
 
+func (c wireFormatConnection) Start() {
+	c.next.Start()
+}
+
 func (c wireFormatConnection) ID() DeviceID {
 	return c.next.ID()
 }

internal/model/model.go

@@ -983,6 +983,8 @@ func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection)
 	}
 	m.rawConn[deviceID] = rawConn
 
+	protoConn.Start()
+
 	cm := m.clusterConfig(deviceID)
 	protoConn.ClusterConfig(cm)
 

internal/model/model_test.go

@@ -242,6 +242,9 @@ func (FakeConnection) Close() error {
 	return nil
 }
 
+func (f FakeConnection) Start() {
+}
+
 func (f FakeConnection) ID() protocol.DeviceID {
 	return f.id
 }