add pending txn subscription support (#3885)

core/events.go

@@ -22,7 +22,7 @@ import (
 )
 
 // NewTxsEvent is posted when a batch of transactions enter the transaction pool.
-type NewTxsEvent struct{ Txs []*types.Transaction }
+type NewTxsEvent struct{ Txs []types.PoolTransaction }
 
 // PendingLogsEvent is posted pre mining and notifies of pending logs.
 type PendingLogsEvent struct {

core/tx_pool.go

@@ -1290,7 +1290,7 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 // invalidated transactions (low nonce, low balance) are deleted.
 func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 	// Track the promoted transactions to broadcast them at once
-	var promoted types.PoolTransactions
+	var promoted []types.PoolTransaction
 	logger := utils.Logger().With().Stack().Logger()
 
 	// Gather all the accounts potentially needing updates
@@ -1353,9 +1353,9 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 		}
 	}
 	// Notify subsystem for new promoted transactions.
-	//if len(promoted) > 0 {
+	if len(promoted) > 0 {
-	//	go pool.txFeed.Send(NewTxsEvent{promoted})
+		go pool.txFeed.Send(NewTxsEvent{promoted})
-	//}
+	}
 	// If the pending limit is overflown, start equalizing allowances
 	pending := uint64(0)
 	for _, list := range pool.pending {

rpc/filters/filter_system.go

@@ -348,7 +348,7 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
 	case core.NewTxsEvent:
 		hashes := make([]common.Hash, 0, len(e.Txs))
 		for _, tx := range e.Txs {
-			hashes = append(hashes, tx.HashByType())
+			hashes = append(hashes, tx.Hash())
 		}
 		for _, f := range filters[PendingTransactionsSubscription] {
 			f.hashes <- hashes