diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index 7a7eb76c20..ff319df41e 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -91,8 +91,21 @@ type EventSystem struct { backend Backend lightMode bool lastHead *types.Header - install chan *subscription // install filter for event notification - uninstall chan *subscription // remove filter for event notification + + // 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 } } diff --git a/event/event.go b/event/event.go index 20d20d1f57..c933f1e84b 100644 --- a/event/event.go +++ b/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()