Do not reuse transports that have been shut down

* raft_test.go: do not reuse transports which have been shutdown
  (see testRaft_SnapshotRestore etc.)

* Make InmemTransport optionally take a specified address. This is to
  provide compatibility with the other transports. It's necessary on
  other Loopback transports as sometimes we Close() a transport, but
  then wish to recreate a new peer. As the transport has been Close()d
  we can't reuse the same transport with the new peer, so we need
  to create a fresh transport with a specified address (being the
  address of the old disconnected peer). This is in any case a better
  test.

Signed-off-by: Alex Bligh <alex@alex.org.uk>

inmem_transport.go

@@ -44,9 +44,11 @@ type InmemTransport struct {
 }
 
 // NewInmemTransport is used to initialize a new transport
-// and generates a random local address.
-func NewInmemTransport() (string, *InmemTransport) {
-	addr := NewInmemAddr()
+// and generates a random local address if none is specified
+func NewInmemTransport(addr string) (string, *InmemTransport) {
+	if addr == "" {
+		addr = NewInmemAddr()
+	}
 	trans := &InmemTransport{
 		consumerCh: make(chan RPC, 16),
 		localAddr:  addr,

peer_test.go

@@ -15,7 +15,7 @@ func TestJSONPeers(t *testing.T) {
 	defer os.RemoveAll(dir)
 
 	// Create the store
-	_, trans := NewInmemTransport()
+	_, trans := NewInmemTransport("")
 	store := NewJSONPeers(dir, trans)
 
 	// Try a read, should get nothing

raft_test.go

@@ -346,7 +346,7 @@ func MakeCluster(n int, t *testing.T, conf *Config) *cluster {
 		c.dirs = append(c.dirs, dir2)
 		c.snaps = append(c.snaps, snap)
 
-		addr, trans := NewInmemTransport()
+		addr, trans := NewInmemTransport("")
 		c.trans = append(c.trans, trans)
 		peers = append(peers, addr)
 	}
@@ -397,7 +397,7 @@ func MakeClusterNoPeers(n int, t *testing.T, conf *Config) *cluster {
 		c.dirs = append(c.dirs, dir2)
 		c.snaps = append(c.snaps, snap)
 
-		_, trans := NewInmemTransport()
+		_, trans := NewInmemTransport("")
 		c.trans = append(c.trans, trans)
 	}
 
@@ -1075,8 +1075,10 @@ func TestRaft_SnapshotRestore(t *testing.T) {
 
 	// Restart the Raft
 	r := leader
+	// Can't just reuse the old transport as it will be closed
+	_, trans2 := NewInmemTransport(r.trans.LocalAddr())
 	r, err := NewRaft(r.conf, r.fsm, r.logs, r.stable,
-		r.snapshots, r.peerStore, r.trans)
+		r.snapshots, r.peerStore, trans2)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
@@ -1139,8 +1141,10 @@ func TestRaft_SnapshotRestore_PeerChange(t *testing.T) {
 	// Restart the Raft with new peers
 	r := leader
 	peerStore := &StaticPeers{StaticPeers: peers}
+	// Can't just reuse the old transport as it will be closed
+	_, trans2 := NewInmemTransport(r.trans.LocalAddr())
 	r, err := NewRaft(r.conf, r.fsm, r.logs, r.stable,
-		r.snapshots, peerStore, r.trans)
+		r.snapshots, peerStore, trans2)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}

util_test.go

@@ -153,7 +153,7 @@ func TestAddUniquePeer(t *testing.T) {
 
 func TestEncodeDecodePeers(t *testing.T) {
 	peers := []string{NewInmemAddr(), NewInmemAddr(), NewInmemAddr()}
-	_, trans := NewInmemTransport()
+	_, trans := NewInmemTransport("")
 
 	// Try to encode/decode
 	buf := encodePeers(peers, trans)