Add a way to signal when a Channel has closed (#718)

Channel has Close() and Closed() methods that tells a Channel to close
and returns whether the Channel is already closed, respectively.  This
adds a CloseChan() that returns a channel that will close once the
Channel has completed closing.

channel.go

@@ -157,6 +157,7 @@ type Channel struct {
 	relayTimerVerify    bool
 	handler             Handler
 	onPeerStatusChanged func(*Peer)
+	closed              chan struct{}
 
 	// mutable contains all the members of Channel which are mutable.
 	mutable struct {
@@ -257,6 +258,7 @@ func NewChannel(serviceName string, opts *ChannelOptions) (*Channel, error) {
 		relayHost:         opts.RelayHost,
 		relayMaxTimeout:   validateRelayMaxTimeout(opts.RelayMaxTimeout, logger),
 		relayTimerVerify:  opts.RelayTimerVerification,
+		closed:            make(chan struct{}),
 	}
 	ch.peers = newRootPeerList(ch, opts.OnPeerStatusChanged).newChild()
 
@@ -758,6 +760,8 @@ func (ch *Channel) connectionCloseStateChange(c *Connection) {
 
 func (ch *Channel) onClosed() {
 	removeClosedChannel(ch)
+
+	close(ch.closed)
 	ch.log.Infof("Channel closed.")
 }
 
@@ -766,6 +770,12 @@ func (ch *Channel) Closed() bool {
 	return ch.State() == ChannelClosed
 }
 
+// ClosedChan returns a channel that will close when the Channel has completely
+// closed.
+func (ch *Channel) ClosedChan() <-chan struct{} {
+	return ch.closed
+}
+
 // State returns the current channel state.
 func (ch *Channel) State() ChannelState {
 	ch.mutable.RLock()
@@ -784,24 +794,31 @@ func (ch *Channel) Close() {
 	var connections []*Connection
 	var channelClosed bool
 
-	ch.mutable.Lock()
+	func() {
+		ch.mutable.Lock()
+		defer ch.mutable.Unlock()
 
-	if ch.mutable.l != nil {
-		ch.mutable.l.Close()
-	}
+		if ch.mutable.state == ChannelClosed {
+			ch.Logger().Info("Channel already closed, skipping additional Close() calls")
+			return
+		}
 
-	// Stop the idle connections timer.
-	ch.mutable.idleSweep.Stop()
+		if ch.mutable.l != nil {
+			ch.mutable.l.Close()
+		}
 
-	ch.mutable.state = ChannelStartClose
-	if len(ch.mutable.conns) == 0 {
-		ch.mutable.state = ChannelClosed
-		channelClosed = true
-	}
-	for _, c := range ch.mutable.conns {
-		connections = append(connections, c)
-	}
-	ch.mutable.Unlock()
+		// Stop the idle connections timer.
+		ch.mutable.idleSweep.Stop()
+
+		ch.mutable.state = ChannelStartClose
+		if len(ch.mutable.conns) == 0 {
+			ch.mutable.state = ChannelClosed
+			channelClosed = true
+		}
+		for _, c := range ch.mutable.conns {
+			connections = append(connections, c)
+		}
+	}()
 
 	for _, c := range connections {
 		c.close(LogField{"reason", "channel closing"})

testutils/test_server.go

@@ -23,7 +23,6 @@ package testutils
 import (
 	"encoding/json"
 	"fmt"
-	"runtime"
 	"strings"
 	"testing"
 	"time"
@@ -293,7 +292,16 @@ func (ts *TestServer) addChannel(createChannel func(t testing.TB, opts *ChannelO
 // it waits for the channels to close.
 func (ts *TestServer) close(ch *tchannel.Channel) {
 	ch.Close()
-	ts.waitForChannelClose(ch)
+
+	select {
+	case <-time.After(Timeout(200 * time.Millisecond)):
+		ts.Errorf("Channel %p did not close after 200 ms, last state: %v", ch, ch.State())
+
+		// The introspected state might help debug why the channel isn't closing.
+		introspected := ch.IntrospectState(&tchannel.IntrospectionOptions{IncludeExchanges: true, IncludeTombstones: true})
+		ts.Logf("Introspected state: %s", spew.Sdump(introspected))
+	case <-ch.ClosedChan():
+	}
 }
 
 func (ts *TestServer) verify(ch *tchannel.Channel) {
@@ -319,36 +327,6 @@ func (ts *TestServer) post() {
 	}
 }
 
-func (ts *TestServer) waitForChannelClose(ch *tchannel.Channel) {
-	if ts.Failed() {
-		return
-	}
-	started := time.Now()
-
-	var state tchannel.ChannelState
-	for i := 0; i < 60; i++ {
-		if state = ch.State(); state == tchannel.ChannelClosed {
-			return
-		}
-
-		runtime.Gosched()
-		if i < 5 {
-			continue
-		}
-
-		sleepFor := time.Duration(i) * 100 * time.Microsecond
-		time.Sleep(Timeout(sleepFor))
-	}
-
-	// Channel is not closing, fail the test.
-	sinceStart := time.Since(started)
-	ts.Errorf("Channel %p did not close after %v, last state: %v", ch, sinceStart, state)
-
-	// The introspected state might help debug why the channel isn't closing.
-	introspected := ch.IntrospectState(&tchannel.IntrospectionOptions{IncludeExchanges: true, IncludeTombstones: true})
-	ts.Logf("Introspected state: %s", spew.Sdump(introspected))
-}
-
 func (ts *TestServer) verifyNoStateLeak(ch *tchannel.Channel) {
 	initial := ts.channelStates[ch]
 	final := comparableState(ch, ts.introspectOpts)