rafthttp: add "ActivePeers" to "Transport"

Signed-off-by: Gyuho Lee <gyuhox@gmail.com>

rafthttp/http_test.go

@@ -387,6 +387,7 @@ func (pr *fakePeer) sendSnap(m raftsnap.Message) {
 func (pr *fakePeer) update(urls types.URLs)                { pr.peerURLs = urls }
 func (pr *fakePeer) attachOutgoingConn(conn *outgoingConn) { pr.connc <- conn }
 func (pr *fakePeer) activeSince() time.Time                { return time.Time{} }
+func (pr *fakePeer) isActive() bool                        { return true }
 func (pr *fakePeer) stop()                                 {}
 func (pr *fakePeer) Pause()                                { pr.paused = true }
 func (pr *fakePeer) Resume()                               { pr.paused = false }

rafthttp/peer.go

@@ -76,6 +76,9 @@ type Peer interface {
 	// activeSince returns the time that the connection with the
 	// peer becomes active.
 	activeSince() time.Time
+	// isActive returns true if the connection to this peer
+	// has been established
+	isActive() bool
 	// stop performs any necessary finalization and terminates the peer
 	// elegantly.
 	stop()
@@ -258,6 +261,8 @@ func (p *peer) attachOutgoingConn(conn *outgoingConn) {
 
 func (p *peer) activeSince() time.Time { return p.status.activeSince() }
 
+func (p *peer) isActive() bool { return p.status.isActive() }
+
 // Pause pauses the peer. The peer will simply drops all incoming
 // messages without returning an error.
 func (p *peer) Pause() {

rafthttp/transport.go

@@ -375,6 +375,18 @@ func (t *Transport) Resume() {
 	}
 }
 
+// ActivePeers returns a channel that closes when an initial
+// peer connection has been established. Use this to wait until the
+// first peer connection becomes active.
+func (t *Transport) ActivePeers() (cnt int) {
+	for _, p := range t.peers {
+		if p.isActive() {
+			cnt++
+		}
+	}
+	return cnt
+}
+
 type nopTransporter struct{}
 
 func NewNopTransporter() Transporter {