fix: confirm block import notifier is closed properly

dot/digest/digest.go

@@ -74,6 +74,8 @@ type resume struct {
 
 // NewHandler returns a new Handler
 func NewHandler(blockState BlockState, epochState EpochState, grandpaState GrandpaState) (*Handler, error) {
+	// TODO (ed) imported channel is unregistered and closed in handler.Stop
+	//
 	imported := make(chan *types.Block, 16)
 	finalised := make(chan *types.FinalisationInfo, 16)
 	iid, err := blockState.RegisterImportedChannel(imported)

dot/rpc/subscription/listeners.go

@@ -123,12 +123,25 @@ type BlockListener struct {
 	cancelTimeout time.Duration
 }
 
+// NewBlockListener constructor for creating BlockListener
+func NewBlockListener(conn *WSConn) *BlockListener {
+	bl := &BlockListener{
+		Channel:       make(chan *types.Block, DEFAULT_BUFFER_SIZE),
+		wsconn:        conn,
+		cancel:        make(chan struct{}, 1),
+		cancelTimeout: defaultCancelTimeout,
+		done:          make(chan struct{}, 1),
+	}
+	return bl
+}
+
 // Listen implementation of Listen interface to listen for importedChan changes
 func (l *BlockListener) Listen() {
 	go func() {
 		defer func() {
 			l.wsconn.BlockAPI.UnregisterImportedChannel(l.ChanID)
 			close(l.done)
+			close(l.Channel)
 		}()
 
 		for {
@@ -180,6 +193,7 @@ func (l *BlockFinalizedListener) Listen() {
 		defer func() {
 			l.wsconn.BlockAPI.UnregisterFinalisedChannel(l.chanID)
 			close(l.done)
+			close(l.channel)
 		}()
 
 		for {
@@ -228,6 +242,20 @@ type ExtrinsicSubmitListener struct {
 	cancelTimeout   time.Duration
 }
 
+// NewExtrinsicSubmitListener constructor to build new ExtrinsicSubmitListener
+func NewExtrinsicSubmitListener(conn *WSConn, extBytes []byte) *ExtrinsicSubmitListener {
+	esl := &ExtrinsicSubmitListener{
+		importedChan:  make(chan *types.Block, DEFAULT_BUFFER_SIZE),
+		wsconn:        conn,
+		extrinsic:     types.Extrinsic(extBytes),
+		finalisedChan: make(chan *types.FinalisationInfo),
+		cancel:        make(chan struct{}, 1),
+		done:          make(chan struct{}, 1),
+		cancelTimeout: defaultCancelTimeout,
+	}
+	return esl
+}
+
 // Listen implementation of Listen interface to listen for importedChan changes
 func (l *ExtrinsicSubmitListener) Listen() {
 	// listen for imported blocks with extrinsic
@@ -236,6 +264,8 @@ func (l *ExtrinsicSubmitListener) Listen() {
 			l.wsconn.BlockAPI.UnregisterImportedChannel(l.importedChanID)
 			l.wsconn.BlockAPI.UnregisterFinalisedChannel(l.finalisedChanID)
 			close(l.done)
+			close(l.importedChan)
+			close(l.finalisedChan)
 		}()
 
 		for {

dot/rpc/subscription/websocket.go

@@ -213,13 +213,9 @@ func (c *WSConn) initStorageChangeListener(reqID float64, params interface{}) (L
 }
 
 func (c *WSConn) initBlockListener(reqID float64, _ interface{}) (Listener, error) {
-	bl := &BlockListener{
-		Channel:       make(chan *types.Block, DEFAULT_BUFFER_SIZE),
-		wsconn:        c,
-		cancel:        make(chan struct{}, 1),
-		cancelTimeout: defaultCancelTimeout,
-		done:          make(chan struct{}, 1),
-	}
+	// TODO (ed) here the imported Channel get unregistered and closed in http.Stop
+	//  there doesn't seem to be an un-subscribe for this
+	bl := NewBlockListener(c)
 
 	if c.BlockAPI == nil {
 		c.safeSendError(reqID, nil, "error BlockAPI not set")
@@ -285,16 +281,9 @@ func (c *WSConn) initExtrinsicWatch(reqID float64, params interface{}) (Listener
 		return nil, err
 	}
 
+	// TODO (ed) the importedChan get deleted by unregister imported channel in defer in Listen (listeners.go line 236)
 	// listen for built blocks
-	esl := &ExtrinsicSubmitListener{
-		importedChan:  make(chan *types.Block, DEFAULT_BUFFER_SIZE),
-		wsconn:        c,
-		extrinsic:     types.Extrinsic(extBytes),
-		finalisedChan: make(chan *types.FinalisationInfo),
-		cancel:        make(chan struct{}, 1),
-		done:          make(chan struct{}, 1),
-		cancelTimeout: defaultCancelTimeout,
-	}
+	esl := NewExtrinsicSubmitListener(c, extBytes)
 
 	if c.BlockAPI == nil {
 		return nil, fmt.Errorf("error BlockAPI not set")

lib/grandpa/grandpa.go

@@ -377,6 +377,7 @@ func (s *Service) initiate() error {
 
 func (s *Service) waitForFirstBlock() error {
 	ch := make(chan *types.Block)
+	// TODO (ed) import channel is unregistered by defer, doesn't seem to be closed
 	id, err := s.blockState.RegisterImportedChannel(ch)
 	if err != nil {
 		return err

lib/grandpa/message_tracker.go

@@ -37,6 +37,7 @@ type tracker struct {
 
 func newTracker(bs BlockState, out chan<- *networkVoteMessage) (*tracker, error) {
 	in := make(chan *types.Block, 16)
+	// TODO (ed) import channel is unregistered and closed in tracker.stop()
 	id, err := bs.RegisterImportedChannel(in)
 	if err != nil {
 		return nil, err