Close the transport on shutdown

When a peer shuts down, close the transport. This needs to be done in
the future (i.e. when all the routines have ended), else the remove
peer does not propagate correctly - this was an issue with pull
request hashicorp#72.

Unfortunately we cannot simply add a Close() to the Transport
interface as we would break back compatibility. Therefore
add a type assertion to ensure that the interface supports
Close before closing it.

Add a Close() method to InmemTransport so this actually gets
tested. InmemTransport has nothing much to close (though it
does call DisconnectAll), but other transports may do.

Signed-off-by: Alex Bligh <[email protected]>

future.go

@@ -100,6 +100,9 @@ func (s *shutdownFuture) Error() error {
 	for s.raft.getRoutines() > 0 {
 		time.Sleep(5 * time.Millisecond)
 	}
+	if closeable, ok := s.raft.trans.(WithClose); ok {
+		closeable.Close()
+	}
 	return nil
 }
 

inmem_transport.go

@@ -208,6 +208,12 @@ func (i *InmemTransport) DisconnectAll() {
 	i.pipelines = nil
 }
 
+// Close is used to permanently disable the transport
+func (i *InmemTransport) Close() error {
+	i.DisconnectAll()
+	return nil
+}
+
 func newInmemPipeline(trans *InmemTransport, peer *InmemTransport, addr string) *inmemPipeline {
 	i := &inmemPipeline{
 		trans:        trans,

transport.go

@@ -60,6 +60,13 @@ type Transport interface {
 	SetHeartbeatHandler(cb func(rpc RPC))
 }
 
+// Close() lives in a separate interface as unfortunately it wasn't in the
+// original interface specification.
+type WithClose interface {
+	// Permanently close a transport, stop all go-routines etc
+	Close() error
+}
+
 // AppendPipeline is used for pipelining AppendEntries requests. It is used
 // to increase the replication throughput by masking latency and better
 // utilizing bandwidth.