*: use queuedTxFeed in SubscribeNewTxsEvent

Author: yysung1123

core/txpool/blobpool/blobpool.go

@@ -1744,7 +1744,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
 		return err
 	}
 
-	go p.queuedTxFeed.Send(core.NewQueuedTxsEvent{Txs: types.Transactions{tx}})
+	go p.queuedTxFeed.Send(core.NewTxsEvent{Txs: types.Transactions{tx}})
 
 	// If the address is not yet known, request exclusivity to track the account
 	// only by this subpool until all transactions are evicted
@@ -2091,7 +2091,7 @@ func (p *BlobPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool
 }
 
 // SubscribeQueuedTransactions subscribes to new queued transaction events.
-func (p *BlobPool) SubscribeQueuedTransactions(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
+func (p *BlobPool) SubscribeQueuedTransactions(ch chan<- core.NewTxsEvent) event.Subscription {
 	return p.queuedTxFeed.Subscribe(ch)
 }
 

core/txpool/legacypool/legacypool.go

@@ -407,7 +407,7 @@ func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs
 }
 
 // SubscribeQueuedTransactions subscribes to new queued transaction events.
-func (pool *LegacyPool) SubscribeQueuedTransactions(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
+func (pool *LegacyPool) SubscribeQueuedTransactions(ch chan<- core.NewTxsEvent) event.Subscription {
 	return pool.queuedTxFeed.Subscribe(ch)
 }
 
@@ -678,7 +678,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
 	from, _ := types.Sender(pool.signer, tx)
 
 	// Broadcast a new tx anyway if it's valid
-	go pool.queuedTxFeed.Send(core.NewQueuedTxsEvent{Txs: types.Transactions{tx}})
+	go pool.queuedTxFeed.Send(core.NewTxsEvent{Txs: types.Transactions{tx}})
 
 	// If the address is not yet known, request exclusivity to track the account
 	// only by this subpool until all transactions are evicted

core/txpool/legacypool/legacypool_test.go

@@ -322,7 +322,7 @@ func validateEvents(events chan core.NewTxsEvent, count int) error {
 
 // validateQueuedEvents checks that the correct number of transaction addition events
 // were fired on the pool's event feed.
-func validateQueuedEvents(events chan core.NewQueuedTxsEvent, count int) error {
+func validateQueuedEvents(events chan core.NewTxsEvent, count int) error {
 	var received []*types.Transaction
 
 	for len(received) < count {
@@ -481,7 +481,7 @@ func TestSubscribePendingAndQueuedTransactions(t *testing.T) {
 	defer pool.Close()
 
 	// Keep listening for authenticated transactions.
-	events := make(chan core.NewQueuedTxsEvent, 32)
+	events := make(chan core.NewTxsEvent, 32)
 	sub := pool.SubscribeQueuedTransactions(events)
 	defer sub.Unsubscribe()
 

core/txpool/subpool.go

@@ -159,7 +159,7 @@ type SubPool interface {
 	SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription
 
 	// SubscribeQueuedTransactions subscribes to new queued transaction events.
-	SubscribeQueuedTransactions(ch chan<- core.NewQueuedTxsEvent) event.Subscription
+	SubscribeQueuedTransactions(ch chan<- core.NewTxsEvent) event.Subscription
 
 	// Nonce returns the next nonce of an account, with all transactions executable
 	// by the pool already applied on top.

core/txpool/txpool.go

@@ -385,8 +385,9 @@ func (p *TxPool) SubscribePendingLocalTxsEvent(ch chan<- core.PendingLocalTxsEve
 	return p.subs.Track(event.JoinSubscriptions(subs...))
 }
 
-// SubscribeNewQueuedTxsEvent registers a subscription of NewQueuedTxsEvent and
-func (p *TxPool) SubscribeNewQueuedTxsEvent(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
+// SubscribeNewQueuedTxsEvent registers a subscription of NewTxsEvent and
+// starts sending event to the given channel.
+func (p *TxPool) SubscribeNewQueuedTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
 	subs := make([]event.Subscription, len(p.subpools))
 	for i, subpool := range p.subpools {
 		subs[i] = subpool.SubscribeQueuedTransactions(ch)

eth/api_backend.go

@@ -400,10 +400,6 @@ func (b *EthAPIBackend) TxPool() *txpool.TxPool {
 }
 
 func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
-	return b.eth.txPool.SubscribeTransactions(ch, true)
-}
-
-func (b *EthAPIBackend) SubscribeNewQueuedTxsEvent(ch chan<- core.NewQueuedTxsEvent) event.Subscription {
 	return b.eth.txPool.SubscribeNewQueuedTxsEvent(ch)
 }
 

eth/filters/api.go

@@ -205,75 +205,6 @@ func (api *FilterAPI) NewPendingTransactions(ctx context.Context, fullTx *bool)
 	return rpcSub, nil
 }
 
-// NewQueuedTransactionFilter creates a filter that fetches queued transaction hashes
-// as transactions enter the pending state.
-//
-// It is part of the filter package because this filter can be used through the
-// `eth_getFilterChanges` polling method that is also used for log filters.
-func (api *FilterAPI) NewQueuedTransactionFilter() rpc.ID {
-	var (
-		queuedTxs   = make(chan []*types.Transaction)
-		queuedTxSub = api.events.SubscribeQueuedTxs(queuedTxs)
-		deadline    = 5 * time.Minute
-	)
-
-	api.filtersMu.Lock()
-	api.filters[queuedTxSub.ID] = &filter{typ: QueuedTransactionsSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: queuedTxSub}
-	api.filtersMu.Unlock()
-
-	go func() {
-		for {
-			select {
-			case ph := <-queuedTxs:
-				api.filtersMu.Lock()
-				if f, found := api.filters[queuedTxSub.ID]; found {
-					f.txs = append(f.txs, ph...)
-				}
-				api.filtersMu.Unlock()
-			case <-queuedTxSub.Err():
-				api.filtersMu.Lock()
-				delete(api.filters, queuedTxSub.ID)
-				api.filtersMu.Unlock()
-				return
-			}
-		}
-	}()
-
-	return queuedTxSub.ID
-}
-
-// NewQueuedTransactions creates a subscription that is triggered each time a transaction
-// enters the transaction pool and was signed from one of the transactions this nodes manages.
-func (api *FilterAPI) NewQueuedTransactions(ctx context.Context) (*rpc.Subscription, error) {
-	notifier, supported := rpc.NotifierFromContext(ctx)
-	if !supported {
-		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
-	}
-
-	rpcSub := notifier.CreateSubscription()
-
-	go func() {
-		txs := make(chan []*types.Transaction, 128)
-		queuedTxSub := api.events.SubscribeQueuedTxs(txs)
-
-		for {
-			select {
-			case hashes := <-txs:
-				// To keep the original behaviour, send a single tx hash in one notification.
-				// TODO(rjl493456442) Send a batch of tx hashes in one notification
-				for _, h := range hashes {
-					notifier.Notify(rpcSub.ID, h)
-				}
-			case <-rpcSub.Err():
-				queuedTxSub.Unsubscribe()
-				return
-			}
-		}
-	}()
-
-	return rpcSub, nil
-}
-
 // NewBlockFilter creates a filter that fetches blocks that are imported into the chain.
 // It is part of the filter package since polling goes with eth_getFilterChanges.
 func (api *FilterAPI) NewBlockFilter() rpc.ID {
@@ -639,10 +570,6 @@ func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
 				f.txs = nil
 				return hashes, nil
 			}
-		case QueuedTransactionsSubscription:
-			txs := f.txs
-			f.txs = nil
-			return returnTransactions(txs), nil
 		case LogsSubscription:
 			logs := f.logs
 			f.logs = nil
@@ -671,15 +598,6 @@ func returnLogs(logs []*types.Log) []*types.Log {
 	return logs
 }
 
-// returnTxs is a helper that will return an empty transaction array case the given transationcs array is nil,
-// otherwise the given transactions array is returned.
-func returnTransactions(txs []*types.Transaction) []*types.Transaction {
-	if txs == nil {
-		return []*types.Transaction{}
-	}
-	return txs
-}
-
 // UnmarshalJSON sets *args fields with given data.
 func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
 	type input struct {

eth/filters/filter_system.go

@@ -68,7 +68,6 @@ type Backend interface {
 	ChainConfig() *params.ChainConfig
 	HistoryPruningCutoff() uint64
 	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
-	SubscribeNewQueuedTxsEvent(chan<- core.NewQueuedTxsEvent) event.Subscription
 	SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
 	SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
 	SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
@@ -157,9 +156,6 @@ const (
 	// PendingTransactionsSubscription queries for pending transactions entering
 	// the pending state
 	PendingTransactionsSubscription
-	// QueuedTransactionsSubscription queries tx hashes for queued
-	// transactions entering the queued state
-	QueuedTransactionsSubscription
 	// BlocksSubscription queries hashes for blocks that are imported
 	BlocksSubscription
 	// TransactionReceiptsSubscription queries for transaction receipts when transactions are included in blocks
@@ -172,9 +168,6 @@ const (
 	// txChanSize is the size of channel listening to NewTxsEvent.
 	// The number is referenced from the size of tx pool.
 	txChanSize = 4096
-	// queuedTxChanSize is the size of channel listening to NewQueuedTxsEvent.
-	// The number is referenced from the size of tx pool.
-	queuedTxChanSize = 4096
 	// rmLogsChanSize is the size of channel listening to RemovedLogsEvent.
 	rmLogsChanSize = 10
 	// logsChanSize is the size of channel listening to LogsEvent.
@@ -190,8 +183,6 @@ type subscription struct {
 	logsCrit  ethereum.FilterQuery
 	logs      chan []*types.Log
 	txs       chan []*types.Transaction
-	hashes    chan []common.Hash
-	queuedTxs chan []*types.Transaction
 	headers   chan *types.Header
 	receipts  chan []*ReceiptWithTx
 	txHashes  map[common.Hash]bool // contains transaction hashes for transactionReceipts subscription filtering
@@ -206,20 +197,18 @@ type EventSystem struct {
 	sys     *FilterSystem
 
 	// Subscriptions
-	txsSub       event.Subscription // Subscription for new transaction event
-	queuedTxsSub event.Subscription // Subscription for new queued 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
+	txsSub    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
 
 	// Channels
-	install     chan *subscription          // install filter for event notification
-	uninstall   chan *subscription          // remove filter for event notification
-	txsCh       chan core.NewTxsEvent       // Channel to receive new transactions event
-	queuedTxsCh chan core.NewQueuedTxsEvent // Channel to receive new queued transactions 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
+	install   chan *subscription         // install filter for event notification
+	uninstall chan *subscription         // remove filter for event notification
+	txsCh     chan core.NewTxsEvent      // Channel to receive new transactions 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,
@@ -230,26 +219,24 @@ type EventSystem struct {
 // or by stopping the given mux.
 func NewEventSystem(sys *FilterSystem) *EventSystem {
 	m := &EventSystem{
-		sys:         sys,
-		backend:     sys.backend,
-		install:     make(chan *subscription),
-		uninstall:   make(chan *subscription),
-		txsCh:       make(chan core.NewTxsEvent, txChanSize),
-		queuedTxsCh: make(chan core.NewQueuedTxsEvent, queuedTxChanSize),
-		logsCh:      make(chan []*types.Log, logsChanSize),
-		rmLogsCh:    make(chan core.RemovedLogsEvent, rmLogsChanSize),
-		chainCh:     make(chan core.ChainEvent, chainEvChanSize),
+		sys:       sys,
+		backend:   sys.backend,
+		install:   make(chan *subscription),
+		uninstall: make(chan *subscription),
+		txsCh:     make(chan core.NewTxsEvent, txChanSize),
+		logsCh:    make(chan []*types.Log, logsChanSize),
+		rmLogsCh:  make(chan core.RemovedLogsEvent, rmLogsChanSize),
+		chainCh:   make(chan core.ChainEvent, chainEvChanSize),
 	}
 
 	// Subscribe events
 	m.txsSub = m.backend.SubscribeNewTxsEvent(m.txsCh)
-	m.queuedTxsSub = m.backend.SubscribeNewQueuedTxsEvent(m.queuedTxsCh)
 	m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
 	m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
 	m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
 
 	// Make sure none of the subscriptions are empty
-	if m.txsSub == nil || m.queuedTxsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil {
+	if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil {
 		log.Crit("Subscribe for event system failed")
 	}
 
@@ -284,8 +271,6 @@ func (sub *Subscription) Unsubscribe() {
 				break uninstallLoop
 			case <-sub.f.logs:
 			case <-sub.f.txs:
-			case <-sub.f.hashes:
-			case <-sub.f.queuedTxs:
 			case <-sub.f.headers:
 			case <-sub.f.receipts:
 			}
@@ -372,8 +357,6 @@ func (es *EventSystem) subscribeLogs(crit ethereum.FilterQuery, logs chan []*typ
 		created:   time.Now(),
 		logs:      logs,
 		txs:       make(chan []*types.Transaction),
-		hashes:    make(chan []common.Hash),
-		queuedTxs: make(chan []*types.Transaction),
 		headers:   make(chan *types.Header),
 		receipts:  make(chan []*ReceiptWithTx),
 		installed: make(chan struct{}),
@@ -391,8 +374,6 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
 		created:   time.Now(),
 		logs:      make(chan []*types.Log),
 		txs:       make(chan []*types.Transaction),
-		hashes:    make(chan []common.Hash),
-		queuedTxs: make(chan []*types.Transaction),
 		headers:   headers,
 		receipts:  make(chan []*ReceiptWithTx),
 		installed: make(chan struct{}),
@@ -410,26 +391,7 @@ func (es *EventSystem) SubscribePendingTxs(txs chan []*types.Transaction) *Subsc
 		created:   time.Now(),
 		logs:      make(chan []*types.Log),
 		txs:       txs,
-		queuedTxs: make(chan []*types.Transaction),
-		headers:   make(chan *types.Header),
-		installed: make(chan struct{}),
-		err:       make(chan error),
-	}
-	return es.subscribe(sub)
-}
-
-// SubscribeQueuedTxs creates a subscription that writes transaction hashes for
-// transactions that enter the transaction pool.
-func (es *EventSystem) SubscribeQueuedTxs(queuedTxs chan []*types.Transaction) *Subscription {
-	sub := &subscription{
-		id:        rpc.NewID(),
-		typ:       QueuedTransactionsSubscription,
-		created:   time.Now(),
-		logs:      make(chan []*types.Log),
-		hashes:    make(chan []common.Hash),
-		queuedTxs: queuedTxs,
 		headers:   make(chan *types.Header),
-		receipts:  make(chan []*ReceiptWithTx),
 		installed: make(chan struct{}),
 		err:       make(chan error),
 	}
@@ -493,20 +455,11 @@ func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent)
 	}
 }
 
-func (es *EventSystem) handleQueuedTxsEvent(filters filterIndex, ev core.NewQueuedTxsEvent) {
-	txs := make([]*types.Transaction, 0, len(ev.Txs))
-	txs = append(txs, ev.Txs...)
-	for _, f := range filters[QueuedTransactionsSubscription] {
-		f.queuedTxs <- txs
-	}
-}
-
 // eventLoop (un)installs filters and processes mux events.
 func (es *EventSystem) eventLoop() {
 	// Ensure all subscriptions get cleaned up
 	defer func() {
 		es.txsSub.Unsubscribe()
-		es.queuedTxsSub.Unsubscribe()
 		es.logsSub.Unsubscribe()
 		es.rmLogsSub.Unsubscribe()
 		es.chainSub.Unsubscribe()
@@ -521,8 +474,6 @@ func (es *EventSystem) eventLoop() {
 		select {
 		case ev := <-es.txsCh:
 			es.handleTxsEvent(index, ev)
-		case ev := <-es.queuedTxsCh:
-			es.handleQueuedTxsEvent(index, ev)
 		case ev := <-es.logsCh:
 			es.handleLogs(index, ev)
 		case ev := <-es.rmLogsCh:
@@ -541,8 +492,6 @@ func (es *EventSystem) eventLoop() {
 		// System stopped
 		case <-es.txsSub.Err():
 			return
-		case <-es.queuedTxsSub.Err():
-			return
 		case <-es.logsSub.Err():
 			return
 		case <-es.rmLogsSub.Err():