accounts,core,eth/filters,les,light,internal/ethapi: Add from/to filtering of ReturnData

2026-08-19 02:12:23 +00:00 · 2018-05-20 18:19:38 -07:00 · 2018-05-20 18:19:38 -07:00 · a92c854be5
commit a92c854be5
parent 929ebc0b2e
14 changed files with 262 additions and 111 deletions
--- a/accounts/abi/bind/backends/simulated.go
+++ b/accounts/abi/bind/backends/simulated.go
@ -470,7 +470,7 @@ func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscr
 	return fb.bc.SubscribeLogsEvent(ch)
 }

-func (fb *filterBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription {
+func (fb *filterBackend) SubscribeTransactionEvent(ch chan<- []*core.TransactionEvent) event.Subscription {
 	return fb.bc.SubscribeTransactionEvent(ch)
 }

--- a/core/blockchain.go
+++ b/core/blockchain.go
@ -1168,10 +1168,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
 			coalescedLogs = append(coalescedLogs, logs...)
 			blockInsertTimer.UpdateSince(bstart)
 			events = append(events, ChainEvent{block, block.Hash(), logs})
-			for _, txPostEvent := range txPostEvents {
-				events = append(events, txPostEvent)
-			}
-			lastCanon = block
+			events = append(events, txPostEvents)

 			// Only count canonical blocks for GC processing time
 			bc.gcproc += proctime
@ -1253,6 +1250,7 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
 		oldChain     types.Blocks
 		commonBlock  *types.Block
 		deletedTxs   types.Transactions
+		txPostEvents []*TransactionEvent
 		deletedLogs  []*types.Log
 		// collectLogs collects the logs that were generated during the
 		// processing of the block that corresponds with the given hash.
@ -1296,6 +1294,8 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
 		return fmt.Errorf("Invalid new chain")
 	}

+	blockLookup := make(map[common.Hash]*big.Int)
+
 	for {
 		if oldBlock.Hash() == newBlock.Hash() {
 			commonBlock = oldBlock
@ -1306,6 +1306,10 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
 		newChain = append(newChain, newBlock)
 		deletedTxs = append(deletedTxs, oldBlock.Transactions()...)
 		collectLogs(oldBlock.Hash())
+		// save block # of each deleted transaction; will need this to deduce signer for TransactionEvent
+		for _, tx := range oldBlock.Transactions() {
+			blockLookup[tx.Hash()] = oldBlock.Number()
+		}

 		oldBlock, newBlock = bc.GetBlock(oldBlock.ParentHash(), oldBlock.NumberU64()-1), bc.GetBlock(newBlock.ParentHash(), newBlock.NumberU64()-1)
 		if oldBlock == nil {
@ -1341,9 +1345,22 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
 		// When transactions get deleted from the database that means the
 		// receipts that were created in the fork must also be deleted
 		rawdb.DeleteTxLookupEntry(bc.db, tx.Hash())
-		// Let ReturnData subscribers know when a transaction is removed from canonical chain
-		bc.txPostFeed.Send(TransactionEvent{TxHash: tx.Hash(), RetData: &types.ReturnData{TxHash: tx.Hash(), Removed: true}})
+
+		// Construct TransactionEvent for subscribers of txPostFeed
+		from, _ := types.Sender(types.MakeSigner(bc.chainConfig, blockLookup[tx.Hash()]), tx)
+
+		data := types.ReturnData{TxHash: tx.Hash(), Removed: true}
+		txEvent := TransactionEvent{TxHash: tx.Hash(),
+			From:    &from,
+			To:      tx.To(),
+			RetData: &data,
 		}
+		txPostEvents = append(txPostEvents, &txEvent)
+	}
+
+	// Let subscribers know when a transaction is removed from canonical chain
+	bc.txPostFeed.Send(txPostEvents)
+
 	if len(deletedLogs) > 0 {
 		go bc.rmLogsFeed.Send(RemovedLogsEvent{deletedLogs})
 	}
@ -1377,7 +1394,7 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) {
 		case ChainSideEvent:
 			bc.chainSideFeed.Send(ev)

-		case TransactionEvent:
+		case []*TransactionEvent:
 			bc.txPostFeed.Send(ev)
 		}
 	}
@ -1567,7 +1584,7 @@ func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Su
 }

 // SubscribeTransactionEvent registers a subscription of SubscribeTransactionEvent.
-func (bc *BlockChain) SubscribeTransactionEvent(ch chan<- TransactionEvent) event.Subscription {
+func (bc *BlockChain) SubscribeTransactionEvent(ch chan<- []*TransactionEvent) event.Subscription {
 	return bc.scope.Track(bc.txPostFeed.Subscribe(ch))
 }

--- a/core/events.go
+++ b/core/events.go
@ -38,6 +38,8 @@ type NewMinedBlockEvent struct{ Block *types.Block }
 // TransactionEvent is posted when a transaction completes execution
 type TransactionEvent struct {
 	TxHash  common.Hash
+	From    *common.Address
+	To      *common.Address
 	RetData *types.ReturnData
 }

--- a/core/state_processor.go
+++ b/core/state_processor.go
@ -53,7 +53,7 @@ func NewStateProcessor(config *params.ChainConfig, bc *BlockChain, engine consen
 // Process returns the receipts and logs accumulated during the process and
 // returns the amount of gas that was used in the process. If any of the
 // transactions failed to execute due to insufficient gas it will return an error.
-func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, []TransactionEvent, error) {
+func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, []*TransactionEvent, error) {
 	var (
 		receipts     types.Receipts
 		usedGas      = new(uint64)
@ -61,7 +61,8 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg
 		allLogs      []*types.Log
 		gp           = new(GasPool).AddGas(block.GasLimit())
 		retData      *types.ReturnData
-		txPostEvents []TransactionEvent
+		txPostEvents []*TransactionEvent
+		signer       = types.MakeSigner(p.config, block.Number())
 	)
 	// Mutate the the block and state according to any hard-fork specs
 	if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 {
@ -77,7 +78,8 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg
 		receipts = append(receipts, receipt)
 		allLogs = append(allLogs, receipt.Logs...)
 		retData = &types.ReturnData{TxHash: tx.Hash(), Data: data}
-		txPostEvents = append(txPostEvents, TransactionEvent{TxHash: tx.Hash(), RetData: retData})
+		from, _ := types.Sender(signer, tx)
+		txPostEvents = append(txPostEvents, &TransactionEvent{TxHash: tx.Hash(), From: &from, To: tx.To(), RetData: retData})
 	}
 	// Finalize the block, applying any consensus engine specific extras (e.g. block rewards)
 	p.engine.Finalize(p.bc, header, statedb, block.Transactions(), block.Uncles(), receipts)
--- a/core/types.go
+++ b/core/types.go
@ -42,5 +42,5 @@ type Validator interface {
 // of gas used in the process and return an error if any of the internal rules
 // failed.
 type Processor interface {
-	Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, []TransactionEvent, error)
+	Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, []*TransactionEvent, error)
 }
--- a/eth/api_backend.go
+++ b/eth/api_backend.go
@ -152,7 +152,7 @@ func (b *EthAPIBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) e
 	return b.eth.BlockChain().SubscribeChainSideEvent(ch)
 }

-func (b *EthAPIBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription {
+func (b *EthAPIBackend) SubscribeTransactionEvent(ch chan<- []*core.TransactionEvent) event.Subscription {
 	return b.eth.BlockChain().SubscribeTransactionEvent(ch)
 }

--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@ -19,7 +19,6 @@ package filters
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"math/big"
 	"reflect"
@ -232,10 +231,10 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er
 	return rpcSub, nil
 }

-func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID {
+func (api *PublicFilterAPI) NewReturnDataFilter(crit TxFilterCriteria) rpc.ID {
 	var (
-		retCh  = make(chan *types.ReturnData)
-		retSub = api.events.SubscribeReturnData(retCh)
+		retCh  = make(chan []*types.ReturnData)
+		retSub = api.events.SubscribeReturnData(retCh, ethereum.TxFilterQuery(crit))
 	)

 	api.filtersMu.Lock()
@ -246,11 +245,16 @@ func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID {
 		for {
 			select {
 			case retData := <-retCh:
+				if len(retData) > 0 {
 					api.filtersMu.Lock()
-				if f, found := api.filters[retSub.ID]; found {
-					f.retData = append(f.retData, *retData)
-				}
+					f, found := api.filters[retSub.ID]
 					api.filtersMu.Unlock()
+					if found {
+						for _, data := range retData {
+							f.retData = append(f.retData, *data)
+						}
+					}
+				}
 			case <-retSub.Err():
 				api.filtersMu.Lock()
 				delete(api.filters, retSub.ID)
@ -263,7 +267,7 @@ func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID {
 	return retSub.ID
 }

-func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, error) {
+func (api *PublicFilterAPI) ReturnData(ctx context.Context, crit TxFilterCriteria) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)
 	if !supported {
 		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@ -275,8 +279,8 @@ func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription,
 	rpcSub.SetType(rpc.ReturnDataSubscription)

 	go func() {
-		retCh := make(chan *types.ReturnData)
-		retSub := api.events.SubscribeReturnData(retCh)
+		retCh := make(chan []*types.ReturnData)
+		retSub := api.events.SubscribeReturnData(retCh, ethereum.TxFilterQuery(crit))

 		for {
 			select {
@ -289,10 +293,14 @@ func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription,
 					log.Warn(fmt.Sprintf("Received update msg of invalid type %s for ReturnData subscription", reflect.TypeOf(msg).String()))
 				}
 			case retData := <-retCh:
-				if _, has := txListen[retData.TxHash]; has {
+				if len(retData) > 0 {
+					for _, data := range retData {
+						if _, has := txListen[data.TxHash]; has {
 							// tx from client just completed execution--send back return data
 							notifier.Notify(rpcSub.ID, retData)
 						}
+					}
+				}
 			case <-rpcSub.Err():
 				retSub.Unsubscribe()
 				return
@ -344,10 +352,14 @@ func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc
 	return rpcSub, nil
 }

-// FilterCriteria represents a request to create a new filter.
+// FilterCriteria represents a request to create a new log filter.
 // Same as ethereum.FilterQuery but with UnmarshalJSON() method.
 type FilterCriteria ethereum.FilterQuery

+// TxFilterCriteria represents a request to create a new transaction filter.
+// Adds UnmarshallJSON() method to ethereum.TxFilterQuery
+type TxFilterCriteria ethereum.TxFilterQuery
+
 // NewFilter creates a new filter and returns the filter id. It can be
 // used to retrieve logs when the state changes. This method cannot be
 // used to fetch logs that are already stored in the state.
@ -534,6 +546,7 @@ func returnRetData(rdata []types.ReturnData) []types.ReturnData {

 // UnmarshalJSON sets *args fields with given data.
 func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
+	var err error
 	type input struct {
 		From      *rpc.BlockNumber `json:"fromBlock"`
 		ToBlock   *rpc.BlockNumber `json:"toBlock"`
@ -542,7 +555,7 @@ func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
 	}

 	var raw input
-	if err := json.Unmarshal(data, &raw); err != nil {
+	if err = json.Unmarshal(data, &raw); err != nil {
 		return err
 	}

@ -558,27 +571,9 @@ func (args *FilterCriteria) UnmarshalJSON(data []byte) error {

 	if raw.Addresses != nil {
 		// raw.Address can contain a single address or an array of addresses
-		switch rawAddr := raw.Addresses.(type) {
-		case []interface{}:
-			for i, addr := range rawAddr {
-				if strAddr, ok := addr.(string); ok {
-					addr, err := decodeAddress(strAddr)
+		args.Addresses, err = decodeAddresses(raw.Addresses, "query")
 		if err != nil {
-						return fmt.Errorf("invalid address at index %d: %v", i, err)
-					}
-					args.Addresses = append(args.Addresses, addr)
-				} else {
-					return fmt.Errorf("non-string address at index %d", i)
-				}
-			}
-		case string:
-			addr, err := decodeAddress(rawAddr)
-			if err != nil {
-				return fmt.Errorf("invalid address: %v", err)
-			}
-			args.Addresses = []common.Address{addr}
-		default:
-			return errors.New("invalid addresses in query")
+			return err
 		}
 	}

@ -626,6 +621,75 @@ func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
 	return nil
 }

+// UnmarshalJSON sets *args fields with given data.
+func (args *TxFilterCriteria) UnmarshalJSON(data []byte) error {
+	var err error
+	type input struct {
+		From interface{} `json:"from"`
+		To   interface{} `json:"to"`
+	}
+
+	var raw input
+	if err = json.Unmarshal(data, &raw); err != nil {
+		return err
+	}
+
+	args.From = nil
+	args.To = nil
+
+	if raw.From != nil {
+		args.From, err = decodeAddresses(raw.From, "from")
+		if err != nil {
+			return err
+		}
+	}
+
+	if raw.To != nil {
+		args.To, err = decodeAddresses(raw.To, "to")
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// Decodes address field from JSON, which could be either
+// a single common.Address or an array []common.Address.
+// second arg is the name of the field being decoded, for error reporting.
+func decodeAddresses(addresses interface{}, field string) ([]common.Address, error) {
+	decoded := []common.Address{}
+
+	if addresses == nil {
+		return decoded, nil
+	}
+
+	switch rawAddr := addresses.(type) {
+	case []interface{}:
+		for i, addr := range rawAddr {
+			if strAddr, ok := addr.(string); ok {
+				addr, err := decodeAddress(strAddr)
+				if err != nil {
+					return decoded, fmt.Errorf("invalid address at index %d: %v", i, err)
+				}
+				decoded = append(decoded, addr)
+			} else {
+				return decoded, fmt.Errorf("non-string address at index %d", i)
+			}
+		}
+	case string:
+		addr, err := decodeAddress(rawAddr)
+		if err != nil {
+			return decoded, fmt.Errorf("invalid address: %v", err)
+		}
+		decoded = []common.Address{addr}
+	default:
+		return decoded, fmt.Errorf("invalid addresses in %s", field)
+	}
+
+	return decoded, nil
+}
+
 func decodeAddress(s string) (common.Address, error) {
 	b, err := hexutil.Decode(s)
 	if err == nil && len(b) != common.AddressLength {
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@ -37,7 +37,7 @@ type Backend interface {
 	GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)

 	SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
-	SubscribeTransactionEvent(chan<- core.TransactionEvent) event.Subscription
+	SubscribeTransactionEvent(chan<- []*core.TransactionEvent) event.Subscription
 	SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
 	SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
 	SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
@ -276,6 +276,40 @@ Logs:
 	return ret
 }

+func filterTxs(txEvents []*core.TransactionEvent, from []common.Address, to []common.Address) (retData []*types.ReturnData) {
+	for _, ev := range txEvents {
+		// check that From and To fields each match one address in the lists from and to, if specified
+
+		if from != nil {
+			for _, addr := range from {
+				if ev.From != nil && addr == *ev.From {
+					goto checkRecipient
+				}
+			}
+			continue
+		}
+
+	checkRecipient:
+		if to != nil {
+			for _, addr := range to {
+				ev_to := ev.To
+				if ev_to == nil { // contract creation
+					ev_to = &common.Address{}
+				}
+				if addr == *ev_to {
+					goto bothMatch
+				}
+			}
+			continue
+		}
+
+	bothMatch:
+		retData = append(retData, ev.RetData)
+	}
+
+	return retData
+}
+
 func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]common.Hash) bool {
 	if len(addresses) > 0 {
 		var included bool
--- a/eth/filters/filter_system.go
+++ b/eth/filters/filter_system.go
@ -88,7 +88,8 @@ type subscription struct {
 	logs      chan []*types.Log
 	hashes    chan common.Hash
 	headers   chan *types.Header
-	retData   chan *types.ReturnData
+	txCrit    ethereum.TxFilterQuery
+	retData   chan []*types.ReturnData
 	installed chan struct{} // closed when the filter is installed
 	err       chan error    // closed when the filter is uninstalled
 }
@ -113,7 +114,7 @@ type EventSystem struct {
 	install   chan *subscription            // install filter for event notification
 	uninstall chan *subscription            // remove filter for event notification
 	txPreCh   chan core.TxPreEvent          // Channel to receive new pre transaction events
-	txCh      chan core.TransactionEvent // Channel to receive new transaction events
+	txCh      chan []*core.TransactionEvent // Channel to receive new transaction events
 	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
@ -133,7 +134,7 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
 		install:   make(chan *subscription),
 		uninstall: make(chan *subscription),
 		txPreCh:   make(chan core.TxPreEvent, txPreChanSize),
-		txCh:      make(chan core.TransactionEvent, txPostChanSize),
+		txCh:      make(chan []*core.TransactionEvent, txPostChanSize),
 		logsCh:    make(chan []*types.Log, logsChanSize),
 		rmLogsCh:  make(chan core.RemovedLogsEvent, rmLogsChanSize),
 		chainCh:   make(chan core.ChainEvent, chainEvChanSize),
@ -327,11 +328,12 @@ func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscr

 // SubscribeReturnData creates a subscription that captures return data for transactions
 //  executed by a particular rpc client
-func (es *EventSystem) SubscribeReturnData(retCh chan *types.ReturnData) *Subscription {
+func (es *EventSystem) SubscribeReturnData(retCh chan []*types.ReturnData, crit ethereum.TxFilterQuery) *Subscription {
 	sub := &subscription{
 		id:        rpc.NewID(),
 		typ:       ReturnDataSubscription,
 		created:   time.Now(),
+		txCrit:    crit,
 		retData:   retCh,
 		installed: make(chan struct{}),
 		err:       make(chan error),
@ -377,9 +379,13 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
 		for _, f := range filters[PendingTransactionsSubscription] {
 			f.hashes <- e.Tx.Hash()
 		}
-	case core.TransactionEvent:
+	case []*core.TransactionEvent:
+		if len(e) > 0 {
 			for _, f := range filters[ReturnDataSubscription] {
-			f.retData <- e.RetData
+				if matchedTxs := filterTxs(e, f.txCrit.From, f.txCrit.To); len(matchedTxs) > 0 {
+					f.retData <- matchedTxs
+				}
+			}
 		}
 	case core.ChainEvent:
 		for _, f := range filters[BlocksSubscription] {
--- a/eth/filters/filter_system_test.go
+++ b/eth/filters/filter_system_test.go
@ -113,7 +113,7 @@ func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subsc
 	return b.chainFeed.Subscribe(ch)
 }

-func (b *testBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription {
+func (b *testBackend) SubscribeTransactionEvent(ch chan<- []*core.TransactionEvent) event.Subscription {
 	return b.txPostFeed.Subscribe(ch)
 }

@ -287,16 +287,17 @@ func TestReturnDataFilter(t *testing.T) {
 		rmLogsFeed            = new(event.Feed)
 		logsFeed              = new(event.Feed)
 		chainFeed             = new(event.Feed)
-		//firstAddr 		= common.HexToAddress("0x1111111111111111111111111111111111111111")
-		//secondAddr		= common.HexToAddress("0x2222222222222222222222222222222222222222")
-		//thirdAddress	= common.HexToHash("0x3333333333333333333333333333333333333333")
-		//		notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999")  // TODO: will need these for testing of from: field criteria when impl
+		firstFromAddr         = common.HexToAddress("0x1111111111111111111111111111111111111111")
+		secondFromAddr        = common.HexToAddress("0x2222222222222222222222222222222222222222")
+		firstToAddr           = common.HexToAddress("0x3333333333333333333333333333333333333333")
+		secondToAddr          = common.HexToAddress("0x0000000000000000000000000000000000000000") // contract creation
+		notUsedAddr           = common.HexToAddress("0x9999999999999999999999999999999999999999")
 		firstTx               = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
 		secondTx              = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222")
 		thirdTx               = common.HexToHash("0x3333333333333333333333333333333333333333333333333333333333333333")
-		forthTx              = common.HexToHash("0x4444444444444444444444444444444444444444444444444444444444444444")
-		firstRetData         = common.Hex2Bytes("0x535353535353535353535353535353535353535353535353535353")
-		secondRetData        = common.Hex2Bytes("0x8888888888888888888888888888888888888888888888888888888888888888888888888888888888888")
+		forthTx               = common.HexToHash("4444444444444444444444444444444444444444444444444444444444444444")
+		firstRetData          = common.Hex2Bytes("535353535353535353535353535353535353535353535353535353")
+		secondRetData         = common.Hex2Bytes("8888888888888888888888888888888888888888888888888888888888888888888888888888888888888")
 		thirdRetData   []byte = nil // no return data for this tx
 		forthRetData          = common.Hex2Bytes("0x77")
 		backend               = &testBackend{mux, db, 0, txPreFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
@ -306,31 +307,49 @@ func TestReturnDataFilter(t *testing.T) {
 	retData := [...]types.ReturnData{
 		{TxHash: firstTx, Data: firstRetData, Removed: false},
 		{TxHash: secondTx, Data: secondRetData, Removed: false},
-		{TxHash: thirdTx, Data: forthRetData, Removed: false}, // third tx added to canonical chain
-		{TxHash: thirdTx, Data: forthRetData, Removed: true},  //  but later removed due to reorg
-		{TxHash: forthTx, Data: thirdRetData, Removed: false},
+		{TxHash: thirdTx, Data: thirdRetData, Removed: false}, // third tx added to canonical chain
+		{TxHash: forthTx, Data: forthRetData, Removed: false},
+		{TxHash: thirdTx, Data: thirdRetData, Removed: true}, //  third tx removed from canonical chain due to reorg
+	}
+
+	txEvents := [...]*core.TransactionEvent{
+		{TxHash: retData[0].TxHash, RetData: &retData[0], From: &firstFromAddr, To: &firstToAddr},
+		{TxHash: retData[1].TxHash, RetData: &retData[1], From: &secondFromAddr, To: &secondToAddr},
+		{TxHash: retData[2].TxHash, RetData: &retData[2], From: &firstFromAddr, To: &secondToAddr},
+		{TxHash: retData[3].TxHash, RetData: &retData[3], From: &secondFromAddr, To: &firstToAddr},
+		{TxHash: retData[4].TxHash, RetData: &retData[4], From: &firstFromAddr, To: &secondToAddr},
 	}

 	testCases := []struct {
-		expected []*types.ReturnData
+		crit     TxFilterCriteria
+		expected []types.ReturnData
 		id       rpc.ID
 	}{
-		0: {[]*types.ReturnData{&retData[0]}, ""},
-		1: {[]*types.ReturnData{&retData[1]}, ""},
-		2: {[]*types.ReturnData{&retData[2], &retData[3]}, ""},
-		3: {[]*types.ReturnData{&retData[3]}, ""},
+		0:  {TxFilterCriteria{From: []common.Address{secondFromAddr}}, []types.ReturnData{retData[1], retData[3]}, ""},
+		1:  {TxFilterCriteria{From: []common.Address{firstFromAddr}}, []types.ReturnData{retData[0], retData[2], retData[4]}, ""},
+		2:  {TxFilterCriteria{From: []common.Address{notUsedAddr}}, []types.ReturnData{}, ""},
+		3:  {TxFilterCriteria{From: []common.Address{notUsedAddr, secondFromAddr}}, []types.ReturnData{retData[1], retData[3]}, ""},
+		4:  {TxFilterCriteria{From: []common.Address{notUsedAddr, firstFromAddr, secondFromAddr}}, []types.ReturnData{retData[0], retData[1], retData[2], retData[3], retData[4]}, ""},
+		5:  {TxFilterCriteria{To: []common.Address{firstToAddr}}, []types.ReturnData{retData[0], retData[3]}, ""},
+		6:  {TxFilterCriteria{To: []common.Address{secondToAddr}}, []types.ReturnData{retData[1], retData[2], retData[4]}, ""},
+		7:  {TxFilterCriteria{To: []common.Address{firstToAddr, secondToAddr}}, []types.ReturnData{retData[0], retData[1], retData[2], retData[3], retData[4]}, ""},
+		8:  {TxFilterCriteria{To: []common.Address{notUsedAddr}}, []types.ReturnData{}, ""},
+		9:  {TxFilterCriteria{From: []common.Address{firstFromAddr}, To: []common.Address{secondToAddr}}, []types.ReturnData{retData[2], retData[4]}, ""},
+		10: {TxFilterCriteria{From: []common.Address{notUsedAddr}, To: []common.Address{secondToAddr}}, []types.ReturnData{}, ""},
+		11: {TxFilterCriteria{From: []common.Address{secondFromAddr}, To: []common.Address{secondToAddr, notUsedAddr}}, []types.ReturnData{retData[1]}, ""},
+		12: {TxFilterCriteria{}, []types.ReturnData{retData[0], retData[1], retData[2], retData[3], retData[4]}, ""},
 	}

 	// create all filters
 	for i := range testCases {
-		testCases[i].id = api.NewReturnDataFilter()
+		testCases[i].id = api.NewReturnDataFilter(testCases[i].crit)
 	}

 	time.Sleep(1 * time.Second)

-	for _, r := range retData {
-		txPostFeed.Send(core.TransactionEvent{TxHash: r.TxHash, RetData: &r})
-	}
+	txPostFeed.Send([]*core.TransactionEvent{txEvents[0]})
+	txPostFeed.Send(txEvents[1:3])
+	txPostFeed.Send(txEvents[3:5])

 	for i, tt := range testCases {
 		var fetched []types.ReturnData
@ -355,14 +374,16 @@ func TestReturnDataFilter(t *testing.T) {

 		if len(fetched) != len(tt.expected) {
 			t.Errorf("invalid number of return data events for case %d, want %d events, got %d", i, len(tt.expected), len(fetched))
-			return
+			//return
 		}

 		for j := range fetched {
-			if fetched[j].Removed {
-				t.Errorf("expected tx not to be removed for tx %d in case %d", j, i)
+			if fetched[j].Removed && !tt.expected[j].Removed {
+				t.Errorf("expected tx not to be removed for tx 0x%x in case %d", tt.expected[j].TxHash, i)
+			} else if !fetched[j].Removed && tt.expected[j].Removed {
+				t.Errorf("expected tx to be removed for tx 0x%xin case %d", tt.expected[j].TxHash, i)
 			}
-			if !reflect.DeepEqual(fetched[i], tt.expected[i]) {
+			if !reflect.DeepEqual(fetched[j], tt.expected[j]) {
 				t.Errorf("invalid return data for tx %d in case %d", j, i)
 			}
 		}
--- a/interfaces.go
+++ b/interfaces.go
@ -149,6 +149,11 @@ type FilterQuery struct {
 	Topics [][]common.Hash
 }

+type TxFilterQuery struct {
+	From []common.Address // restricts matches to transactions originating from specific addresses
+	To   []common.Address // restricts matches to transactions sent to specific addresses
+}
+
 // LogFilterer provides access to contract log events using a one-off query or continuous
 // event subscription.
 //
--- a/internal/ethapi/backend.go
+++ b/internal/ethapi/backend.go
@ -57,7 +57,7 @@ type Backend interface {
 	SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
 	SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
 	SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription
-	SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription
+	SubscribeTransactionEvent(ch chan<- []*core.TransactionEvent) event.Subscription

 	// TxPool API
 	SendTx(ctx context.Context, signedTx *types.Transaction) error
--- a/les/api_backend.go
+++ b/les/api_backend.go
@ -140,7 +140,7 @@ func (b *LesApiBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Sub
 	return b.eth.txPool.SubscribeTxPreEvent(ch)
 }

-func (b *LesApiBackend) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription {
+func (b *LesApiBackend) SubscribeTransactionEvent(ch chan<- []*core.TransactionEvent) event.Subscription {
 	return b.eth.blockchain.SubscribeTransactionEvent(ch)
 }

--- a/light/lightchain.go
+++ b/light/lightchain.go
@ -500,7 +500,7 @@ func (self *LightChain) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) e
 }

 // SubscribeTransactionEvent registers a subscription of TransactionEvent.
-func (self *LightChain) SubscribeTransactionEvent(ch chan<- core.TransactionEvent) event.Subscription {
+func (self *LightChain) SubscribeTransactionEvent(ch chan<- []*core.TransactionEvent) event.Subscription {
 	return self.scope.Track(self.txPostFeed.Subscribe(ch))
 }