eth/filters, accounts, rpc: abort system if subscribe failed

eth/filters/filter_system.go

@@ -91,8 +91,21 @@ type EventSystem struct {
 	backend   Backend
 	lightMode bool
 	lastHead  *types.Header
+
+	// Subscriptions
+	txSub         event.Subscription         // Subscription for new transaction event
+	logsSub       event.Subscription         // Subscription for new log event
+	rmLogsSub     event.Subscription         // Subscription for removed log event
+	chainSub      event.Subscription         // Subscription for new chain event
+	pendingLogSub *event.TypeMuxSubscription // Subscription for pending log event
+
+	// Channels
 	install   chan *subscription         // install filter for event notification
 	uninstall chan *subscription         // remove filter for event notification
+	txCh      chan core.TxPreEvent       // Channel to receive new transaction event
+	logsCh    chan []*types.Log          // Channel to receive new log event
+	rmLogsCh  chan core.RemovedLogsEvent // Channel to receive removed log event
+	chainCh   chan core.ChainEvent       // Channel to receive new chain event
 }
 
 // NewEventSystem creates a new manager that listens for event on the given mux,
@@ -108,10 +121,36 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
 		lightMode: lightMode,
 		install:   make(chan *subscription),
 		uninstall: make(chan *subscription),
+		txCh:      make(chan core.TxPreEvent, txChanSize),
+		logsCh:    make(chan []*types.Log, logsChanSize),
+		rmLogsCh:  make(chan core.RemovedLogsEvent, rmLogsChanSize),
+		chainCh:   make(chan core.ChainEvent, chainEvChanSize),
+	}
+
+	// Subscribe events
+	m.txSub = m.backend.SubscribeTxPreEvent(m.txCh)
+	if m.txSub == nil {
+		return nil
+	}
+	m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
+	if m.logsSub == nil {
+		return nil
+	}
+	m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
+	if m.rmLogsSub == nil {
+		return nil
+	}
+	m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
+	if m.chainSub == nil {
+		return nil
+	}
+	// TODO(rjl493456442): use feed to subscribe pending log event
+	m.pendingLogSub = m.mux.Subscribe(core.PendingLogsEvent{})
+	if m.pendingLogSub.Closed() {
+		return nil
 	}
 
 	go m.eventLoop()
-
 	return m
 }
 
@@ -411,40 +450,15 @@ func (es *EventSystem) lightFilterLogs(header *types.Header, addresses []common.
 
 // eventLoop (un)installs filters and processes mux events.
 func (es *EventSystem) eventLoop() {
-	var (
-		index = make(filterIndex)
-		sub   = es.mux.Subscribe(core.PendingLogsEvent{})
-		// Subscribe TxPreEvent form txpool
-		txCh  = make(chan core.TxPreEvent, txChanSize)
-		txSub = es.backend.SubscribeTxPreEvent(txCh)
-		// Subscribe RemovedLogsEvent
-		rmLogsCh  = make(chan core.RemovedLogsEvent, rmLogsChanSize)
-		rmLogsSub = es.backend.SubscribeRemovedLogsEvent(rmLogsCh)
-		// Subscribe []*types.Log
-		logsCh  = make(chan []*types.Log, logsChanSize)
-		logsSub = es.backend.SubscribeLogsEvent(logsCh)
-		// Subscribe ChainEvent
-		chainEvCh  = make(chan core.ChainEvent, chainEvChanSize)
-		chainEvSub = es.backend.SubscribeChainEvent(chainEvCh)
-	)
+	var index = make(filterIndex)
 
 	defer func() {
 		// Unsubscribe all events
-		if sub != nil {
-			sub.Unsubscribe()
-		}
-		if txSub != nil {
-			txSub.Unsubscribe()
-		}
-		if rmLogsSub != nil {
-			rmLogsSub.Unsubscribe()
-		}
-		if logsSub != nil {
-			logsSub.Unsubscribe()
-		}
-		if chainEvSub != nil {
-			chainEvSub.Unsubscribe()
-		}
+		es.pendingLogSub.Unsubscribe()
+		es.txSub.Unsubscribe()
+		es.logsSub.Unsubscribe()
+		es.rmLogsSub.Unsubscribe()
+		es.chainSub.Unsubscribe()
 	}()
 
 	for i := UnknownSubscription; i < LastIndexSubscription; i++ {
@@ -453,22 +467,21 @@ func (es *EventSystem) eventLoop() {
 
 	for {
 		select {
-		case ev, active := <-sub.Chan():
+		// Handle subscribed events
+		case ev := <-es.txCh:
+			es.broadcast(index, ev)
+		case ev := <-es.logsCh:
+			es.broadcast(index, ev)
+		case ev := <-es.rmLogsCh:
+			es.broadcast(index, ev)
+		case ev := <-es.chainCh:
+			es.broadcast(index, ev)
+		case ev, active := <-es.pendingLogSub.Chan():
 			if !active { // system stopped
 				return
 			}
 			es.broadcast(index, ev)
 
-		// Handle subscribed events
-		case ev := <-txCh:
-			es.broadcast(index, ev)
-		case ev := <-rmLogsCh:
-			es.broadcast(index, ev)
-		case ev := <-logsCh:
-			es.broadcast(index, ev)
-		case ev := <-chainEvCh:
-			es.broadcast(index, ev)
-
 		case f := <-es.install:
 			if f.typ == MinedAndPendingLogsSubscription {
 				// the type are logs and pending logs subscriptions
@@ -478,6 +491,7 @@ func (es *EventSystem) eventLoop() {
 				index[f.typ][f.id] = f
 			}
 			close(f.installed)
+
 		case f := <-es.uninstall:
 			if f.typ == MinedAndPendingLogsSubscription {
 				// the type are logs and pending logs subscriptions
@@ -489,13 +503,13 @@ func (es *EventSystem) eventLoop() {
 			close(f.err)
 
 		// System stopped
-		case <-txSub.Err():
+		case <-es.txSub.Err():
 			return
-		case <-rmLogsSub.Err():
+		case <-es.logsSub.Err():
 			return
-		case <-logsSub.Err():
+		case <-es.rmLogsSub.Err():
 			return
-		case <-chainEvSub.Err():
+		case <-es.chainSub.Err():
 			return
 		}
 	}

event/event.go

@@ -180,6 +180,10 @@ func (s *TypeMuxSubscription) Unsubscribe() {
 	s.closewait()
 }
 
+func (s *TypeMuxSubscription) Closed() bool {
+	return s.closed
+}
+
 func (s *TypeMuxSubscription) closewait() {
 	s.closeMu.Lock()
 	defer s.closeMu.Unlock()