From 5f548c7679dbc4c9ab0ff6ead0dacc77bbf719db Mon Sep 17 00:00:00 2001
From: Ethan Buchman <ethan@coinculture.info>
Date: Tue, 3 Apr 2018 22:56:46 +0300
Subject: [PATCH] consensus: close pubsub channels. fixes #1372

---
 consensus/reactor.go | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/consensus/reactor.go b/consensus/reactor.go
index b13be0e87..c71f9e0c4 100644
--- a/consensus/reactor.go
+++ b/consensus/reactor.go
@@ -371,19 +371,21 @@ func (conR *ConsensusReactor) startBroadcastRoutine() error {
 	}
 
 	go func() {
+		var data interface{}
+		var ok bool
 		for {
 			select {
-			case data, ok := <-stepsCh:
+			case data, ok = <-stepsCh:
 				if ok { // a receive from a closed channel returns the zero value immediately
 					edrs := data.(types.TMEventData).Unwrap().(types.EventDataRoundState)
 					conR.broadcastNewRoundStep(edrs.RoundState.(*cstypes.RoundState))
 				}
-			case data, ok := <-votesCh:
+			case data, ok = <-votesCh:
 				if ok {
 					edv := data.(types.TMEventData).Unwrap().(types.EventDataVote)
 					conR.broadcastHasVoteMessage(edv.Vote)
 				}
-			case data, ok := <-heartbeatsCh:
+			case data, ok = <-heartbeatsCh:
 				if ok {
 					edph := data.(types.TMEventData).Unwrap().(types.EventDataProposalHeartbeat)
 					conR.broadcastProposalHeartbeatMessage(edph)
@@ -392,6 +394,10 @@ func (conR *ConsensusReactor) startBroadcastRoutine() error {
 				conR.eventBus.UnsubscribeAll(ctx, subscriber)
 				return
 			}
+			if !ok {
+				conR.eventBus.UnsubscribeAll(ctx, subscriber)
+				return
+			}
 		}
 	}()