just subscribe directly using SubscribeChainEvent

eth/api_backend.go

@@ -36,12 +36,10 @@ import (
 	"github.com/ethereum/go-ethereum/core/txpool/locals"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
-	"github.com/ethereum/go-ethereum/eth/filters"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
 	"github.com/ethereum/go-ethereum/eth/tracers"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
-	"github.com/ethereum/go-ethereum/internal/ethapi"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -496,30 +494,3 @@ func (b *EthAPIBackend) RPCTxSyncDefaultTimeout() time.Duration {
 func (b *EthAPIBackend) RPCTxSyncMaxTimeout() time.Duration {
 	return b.eth.config.TxSyncMaxTimeout
 }
-
-func (b *EthAPIBackend) SubscribeTransactionReceipts(
-	txHashes []common.Hash,
-	out chan<- []*ethapi.ReceiptWithTx,
-) event.Subscription {
-	ch := make(chan core.ChainEvent, 16)
-	sub := b.eth.blockchain.SubscribeChainEvent(ch)
-
-	go func() {
-		defer sub.Unsubscribe()
-		for {
-			select {
-			case ev, ok := <-ch:
-				if !ok {
-					return
-				}
-				batch := filters.FilterReceipts(txHashes, ev)
-				if len(batch) > 0 {
-					out <- batch
-				}
-			case <-sub.Err():
-				return
-			}
-		}
-	}()
-	return sub
-}

eth/filters/filter.go

@@ -29,7 +29,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/filtermaps"
 	"github.com/ethereum/go-ethereum/core/history"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/internal/ethapi"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -555,13 +554,16 @@ func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]commo
 }
 
 // ReceiptWithTx contains a receipt and its corresponding transaction
-type ReceiptWithTx = ethapi.ReceiptWithTx
+type ReceiptWithTx struct {
+	Receipt     *types.Receipt
+	Transaction *types.Transaction
+}
 
-// FilterReceipts returns the receipts matching the given criteria
+// filterReceipts returns the receipts matching the given criteria
 // In addition to returning receipts, it also returns the corresponding transactions.
 // This is because receipts only contain low-level data, while user-facing data
 // may require additional information from the Transaction.
-func FilterReceipts(txHashes []common.Hash, ev core.ChainEvent) []*ReceiptWithTx {
+func filterReceipts(txHashes []common.Hash, ev core.ChainEvent) []*ReceiptWithTx {
 	var ret []*ReceiptWithTx
 
 	receipts := ev.Receipts

eth/filters/filter_system.go

@@ -445,7 +445,7 @@ func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent)
 
 	// Handle transaction receipts subscriptions when a new block is added
 	for _, f := range filters[TransactionReceiptsSubscription] {
-		matchedReceipts := FilterReceipts(f.txHashes, ev)
+		matchedReceipts := filterReceipts(f.txHashes, ev)
 		if len(matchedReceipts) > 0 {
 			f.receipts <- matchedReceipts
 		}

internal/ethapi/api.go

@@ -1652,11 +1652,6 @@ func (api *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil
 	return SubmitTransaction(ctx, api.b, tx)
 }
 
-type ReceiptWithTx struct {
-	Receipt     *types.Receipt
-	Transaction *types.Transaction
-}
-
 // SendRawTransactionSync will add the signed transaction to the transaction pool
 // and wait until the transaction has been included in a block and return the receipt, or the timeout.
 func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hexutil.Bytes, timeoutMs *hexutil.Uint64) (map[string]interface{}, error) {
@@ -1664,6 +1659,12 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex
 	if err := tx.UnmarshalBinary(input); err != nil {
 		return nil, err
 	}
+
+	ch := make(chan core.ChainEvent, 128)
+	sub := api.b.SubscribeChainEvent(ch)
+	subErrCh := sub.Err()
+	defer sub.Unsubscribe()
+
 	hash, err := SubmitTransaction(ctx, api.b, tx)
 	if err != nil {
 		return nil, err
@@ -1690,13 +1691,6 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex
 		return r, nil
 	}
 
-	// Subscribe to receipt stream (filtered to this tx)
-	receipts := make(chan []*ReceiptWithTx, 1)
-	sub := api.b.SubscribeTransactionReceipts([]common.Hash{hash}, receipts)
-	defer sub.Unsubscribe()
-
-	subErrCh := sub.Err()
-
 	for {
 		select {
 		case <-receiptCtx.Done():
@@ -1717,20 +1711,23 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex
 			}
 			return nil, err
 
-		case batch := <-receipts:
+		case ev := <-ch:
-			for _, rwt := range batch {
+			rs := ev.Receipts
-				if rwt == nil || rwt.Receipt == nil || rwt.Receipt.TxHash != hash {
+			txs := ev.Transactions
+			if len(rs) == 0 || len(rs) != len(txs) {
 				continue
 			}
-
+			for i := range rs {
-				if rwt.Receipt.BlockNumber != nil && rwt.Receipt.BlockHash != (common.Hash{}) {
+				if rs[i].TxHash == hash {
+					if rs[i].BlockNumber != nil && rs[i].BlockHash != (common.Hash{}) {
+						signer := types.LatestSigner(api.b.ChainConfig())
 						return MarshalReceipt(
-						rwt.Receipt,
+							rs[i],
-						rwt.Receipt.BlockHash,
+							rs[i].BlockHash,
-						rwt.Receipt.BlockNumber.Uint64(),
+							rs[i].BlockNumber.Uint64(),
-						api.signer,
+							signer,
-						rwt.Transaction,
+							txs[i],
-						int(rwt.Receipt.TransactionIndex),
+							int(rs[i].TransactionIndex),
 						), nil
 					}
 					return api.GetTransactionReceipt(receiptCtx, hash)
@@ -1738,6 +1735,7 @@ func (api *TransactionAPI) SendRawTransactionSync(ctx context.Context, input hex
 			}
 		}
 	}
+}
 
 // Sign calculates an ECDSA signature for:
 // keccak256("\x19Ethereum Signed Message:\n" + len(message) + message).

internal/ethapi/api_test.go

@@ -441,7 +441,7 @@ type testBackend struct {
 	pending         *types.Block
 	pendingReceipts types.Receipts
 
-	receiptsFeed *event.Feed
+	chainFeed *event.Feed
 	autoMine  bool
 
 	sentTx     *types.Transaction
@@ -479,7 +479,7 @@ func newTestBackend(t *testing.T, n int, gspec *core.Genesis, engine consensus.E
 		acc:             acc,
 		pending:         blocks[n],
 		pendingReceipts: receipts[n],
-		receiptsFeed:    new(event.Feed),
+		chainFeed:       new(event.Feed),
 	}
 	return backend
 }
@@ -601,7 +601,7 @@ func (b testBackend) GetEVM(ctx context.Context, state *state.StateDB, header *t
 	return vm.NewEVM(context, state, b.chain.Config(), *vmConfig)
 }
 func (b testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
-	panic("implement me")
+	return b.chainFeed.Subscribe(ch)
 }
 func (b testBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
 	panic("implement me")
@@ -610,11 +610,12 @@ func (b *testBackend) SendTx(ctx context.Context, tx *types.Transaction) error {
 	b.sentTx = tx
 	b.sentTxHash = tx.Hash()
 
-	if b.autoMine && b.receiptsFeed != nil {
+	if b.autoMine {
+		// Synthesize a "mined" receipt at head+1
 		num := b.chain.CurrentHeader().Number.Uint64() + 1
-		bh := fakeBlockHash(tx.Hash())
 
-		r := &types.Receipt{
+		bh := fakeBlockHash(tx.Hash())
+		receipt := &types.Receipt{
 			TxHash:            tx.Hash(),
 			Status:            types.ReceiptStatusSuccessful,
 			BlockHash:         bh,
@@ -623,7 +624,16 @@ func (b *testBackend) SendTx(ctx context.Context, tx *types.Transaction) error {
 			CumulativeGasUsed: 21000,
 			GasUsed:           21000,
 		}
-		b.receiptsFeed.Send([]*ReceiptWithTx{{Receipt: r, Transaction: tx}})
+		hdr := &types.Header{
+			Number: new(big.Int).SetUint64(num),
+		}
+
+		// Broadcast a ChainEvent that includes the receipts and txs
+		b.chainFeed.Send(core.ChainEvent{
+			Header:       hdr,
+			Receipts:     types.Receipts{receipt},
+			Transactions: types.Transactions{tx},
+		})
 	}
 	return nil
 }
@@ -3981,13 +3991,6 @@ func makeSelfSignedRaw(t *testing.T, api *TransactionAPI, addr common.Address) (
 	return makeSignedRaw(t, api, addr, addr, big.NewInt(0))
 }
 
-func (b *testBackend) SubscribeTransactionReceipts(_ []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription {
-	if b.receiptsFeed == nil {
-		return event.NewSubscription(func(quit <-chan struct{}) error { <-quit; return nil })
-	}
-	// Test will only publish the receipts it wants; we just forward.
-	return b.receiptsFeed.Subscribe(ch)
-}
 func TestSendRawTransactionSync_Success(t *testing.T) {
 	t.Parallel()
 	genesis := &core.Genesis{

internal/ethapi/backend.go

@@ -98,7 +98,6 @@ type Backend interface {
 	GetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error)
 	SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
 	SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
-	SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription
 
 	CurrentView() *filtermaps.ChainView
 	NewMatcherBackend() filtermaps.MatcherBackend

internal/ethapi/transaction_args_test.go

@@ -411,6 +411,3 @@ func (b *backendMock) CurrentView() *filtermaps.ChainView           { return nil
 func (b *backendMock) NewMatcherBackend() filtermaps.MatcherBackend { return nil }
 
 func (b *backendMock) HistoryPruningCutoff() uint64 { return 0 }
-func (b *backendMock) SubscribeTransactionReceipts(txHashes []common.Hash, ch chan<- []*ReceiptWithTx) event.Subscription {
-	return nil
-}