forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-25 14:16:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/filter: refactor log filter system

Browse source
This commit is contained in:
Bas van Kervel 2016-06-22 09:34:33 +02:00
parent f970610c04
commit f99cc5a840
5 changed files with 1138 additions and 729 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

596
eth/filters/api.go
View file

@ -17,283 +17,191 @@
package filters package filters
import ( import (
"crypto/rand"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"sync" "sync"
"time"
"golang.org/x/net/context"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"golang.org/x/net/context"
) )
var ( // PublicFilterAPI allows clients to retrieve or get notified of data such as blocks,
filterTickerTime = 5 * time.Minute // transactions and logs.
)
// byte will be inferred
const (
unknownFilterTy = iota
blockFilterTy
transactionFilterTy
logFilterTy
)
// PublicFilterAPI offers support to create and manage filters. This will allow external clients to retrieve various
// information related to the Ethereum protocol such als blocks, transactions and logs.
type PublicFilterAPI struct { type PublicFilterAPI struct {
mux *event.TypeMux db ethdb.Database
m *Manager
quit chan struct{} subscriptionsMu sync.Mutex
chainDb ethdb.Database subscriptions map[string]FilterID // mapping between subscriptions and the underlying filters
filterManager *FilterSystem
filterMapMu sync.RWMutex
filterMapping map[string]int // maps between filter internal filter identifiers and external filter identifiers
logMu sync.RWMutex
logQueue map[int]*logQueue
blockMu sync.RWMutex
blockQueue map[int]*hashQueue
transactionMu sync.RWMutex
transactionQueue map[int]*hashQueue
transactMu sync.Mutex
} }
// NewPublicFilterAPI returns a new PublicFilterAPI instance. // NewPublicFilterAPI returns a new PublicFilterAPI instance.
// It uses the given database to retrieve stored logs and the mux to listen for events.
func NewPublicFilterAPI(chainDb ethdb.Database, mux *event.TypeMux) *PublicFilterAPI { func NewPublicFilterAPI(chainDb ethdb.Database, mux *event.TypeMux) *PublicFilterAPI {
svc := &PublicFilterAPI{ return &PublicFilterAPI{
mux: mux, db: chainDb,
chainDb: chainDb, m: NewManager(mux),
filterManager: NewFilterSystem(mux), subscriptions: make(map[string]FilterID),
filterMapping: make(map[string]int),
logQueue: make(map[int]*logQueue),
blockQueue: make(map[int]*hashQueue),
transactionQueue: make(map[int]*hashQueue),
} }
go svc.start()
return svc
} }
// Stop quits the work loop. // NewBlockFilter creates a filter that returns hashes for new blocks.
func (s *PublicFilterAPI) Stop() { // To check if the state has changed, call GetFilterChanges.
close(s.quit) //
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newblockfilter
func (api *PublicFilterAPI) NewBlockFilter() (FilterID, error) {
return api.m.NewBlockFilter()
} }
// start the work loop, wait and process events. // NewPendingTransactionFilter creates a filter, to notify when new pending transactions arrive.
func (s *PublicFilterAPI) start() { // To check if the state has changed, call GetFilterChanges.
timer := time.NewTicker(2 * time.Second) //
defer timer.Stop() // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter
done: func (api *PublicFilterAPI) NewPendingTransactionFilter() (FilterID, error) {
for { return api.m.NewPendingTransactionFilter()
select { }
case <-timer.C:
s.logMu.Lock()
for id, filter := range s.logQueue {
if time.Since(filter.timeout) > filterTickerTime {
s.filterManager.Remove(id)
delete(s.logQueue, id)
}
}
s.logMu.Unlock()
s.blockMu.Lock() // GetFilterChanges returns new logs for the given filter since last time is was called.
for id, filter := range s.blockQueue { // This can be used for polling.
if time.Since(filter.timeout) > filterTickerTime { //
s.filterManager.Remove(id) // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterchanges
delete(s.blockQueue, id) func (api *PublicFilterAPI) GetFilterChanges(id FilterID) (interface{}, error) {
} // filter might be deleted between retrieving filter type and calling the XXXFilterChanges
} // this is not an issue, XXXFilterChanges will return an errFilterNotFound error.
s.blockMu.Unlock() switch api.m.FilterType(id) {
case BlockFilter:
s.transactionMu.Lock() hashes, err := api.m.GetBlockFilterChanges(id)
for id, filter := range s.transactionQueue { return toRPCHashes(hashes), err
if time.Since(filter.timeout) > filterTickerTime { case PendingTxFilter:
s.filterManager.Remove(id) hashes, err := api.m.GetPendingTxFilterChanges(id)
delete(s.transactionQueue, id) return toRPCHashes(hashes), err
} case LogFilter:
} logs, err := api.m.GetLogFilterChanges(id)
s.transactionMu.Unlock() return toRPCLogs(logs), err
case <-s.quit:
break done
}
} }
return nil, errFilterNotFound
} }
// NewBlockFilter create a new filter that returns blocks that are included into the canonical chain. // NewFilter creates a filter that can be used to fetch new logs.
func (s *PublicFilterAPI) NewBlockFilter() (string, error) { // Note: as specification dictates, it can be used to fetch logs when the state changes.
externalId, err := newFilterId() // Older already created logs cannot be fetched using this method, use GetLogs.
// Therefore the fromBlock and toBlock parameters are ignored.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newfilter
func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (FilterID, error) {
return api.m.NewLogFilter(crit, nil)
}
// GetLogs returns logs matching the given criteria.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs
func (api *PublicFilterAPI) GetLogs(crit FilterCriteria) []Log {
filter := New(api.db)
filter.SetBeginBlock(crit.FromBlock.Int64())
filter.SetEndBlock(crit.ToBlock.Int64())
filter.SetAddresses(crit.Addresses)
filter.SetTopics(crit.Topics)
return toRPCLogs(filter.Find())
}
// GetFilterLogs return all logs that match for the given (log) filter.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterlogs
func (api *PublicFilterAPI) GetFilterLogs(id FilterID) ([]Log, error) {
crit, err := api.m.GetLogFilterCriteria(id)
if err != nil { if err != nil {
return "", err return nil, err
} }
return api.GetLogs(crit), nil
s.blockMu.Lock()
filter := New(s.chainDb)
id, err := s.filterManager.Add(filter, ChainFilter)
if err != nil {
return "", err
}
s.blockQueue[id] = &hashQueue{timeout: time.Now()}
filter.BlockCallback = func(block *types.Block, logs vm.Logs) {
s.blockMu.Lock()
defer s.blockMu.Unlock()
if queue := s.blockQueue[id]; queue != nil {
queue.add(block.Hash())
}
}
defer s.blockMu.Unlock()
s.filterMapMu.Lock()
s.filterMapping[externalId] = id
s.filterMapMu.Unlock()
return externalId, nil
} }
// NewPendingTransactionFilter creates a filter that returns new pending transactions. // Logs creates a subscription that fires each time a log is created that matches the given criteria.
func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) { // Note, in case of chain reorganisations logs with the Removed true property can be returned indicating
externalId, err := newFilterId() // a previously sent log is reverted.
if err != nil { func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (subscription rpc.Subscription, err error) {
return "", err
}
s.transactionMu.Lock()
defer s.transactionMu.Unlock()
filter := New(s.chainDb)
id, err := s.filterManager.Add(filter, PendingTxFilter)
if err != nil {
return "", err
}
s.transactionQueue[id] = &hashQueue{timeout: time.Now()}
filter.TransactionCallback = func(tx *types.Transaction) {
s.transactionMu.Lock()
defer s.transactionMu.Unlock()
if queue := s.transactionQueue[id]; queue != nil {
queue.add(tx.Hash())
}
}
s.filterMapMu.Lock()
s.filterMapping[externalId] = id
s.filterMapMu.Unlock()
return externalId, nil
}
// newLogFilter creates a new log filter.
func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash, callback func(log *vm.Log, removed bool)) (int, error) {
s.logMu.Lock()
defer s.logMu.Unlock()
filter := New(s.chainDb)
id, err := s.filterManager.Add(filter, LogFilter)
if err != nil {
return 0, err
}
s.logQueue[id] = &logQueue{timeout: time.Now()}
filter.SetBeginBlock(earliest)
filter.SetEndBlock(latest)
filter.SetAddresses(addresses)
filter.SetTopics(topics)
filter.LogCallback = func(log *vm.Log, removed bool) {
if callback != nil {
callback(log, removed)
} else {
s.logMu.Lock()
defer s.logMu.Unlock()
if queue := s.logQueue[id]; queue != nil {
queue.add(vmlog{log, removed})
}
}
}
return id, nil
}
// Logs creates a subscription that fires for all new log that match the given filter criteria.
func (s *PublicFilterAPI) Logs(ctx context.Context, args NewFilterArgs) (rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx) notifier, supported := rpc.NotifierFromContext(ctx)
if !supported { if !supported {
return nil, rpc.ErrNotificationsUnsupported return nil, rpc.ErrNotificationsUnsupported
} }
var ( onUnsubscribe := func(subid string) {
externalId string api.subscriptionsMu.Lock()
subscription rpc.Subscription defer api.subscriptionsMu.Unlock()
err error if fid, found := api.subscriptions[subid]; found {
) delete(api.subscriptions, subid)
api.m.Uninstall(fid) // uninstall associated filter
if externalId, err = newFilterId(); err != nil {
return nil, err
}
// uninstall filter when subscription is unsubscribed/cancelled
if subscription, err = notifier.NewSubscription(func(string) {
s.UninstallFilter(externalId)
}); err != nil {
return nil, err
}
notifySubscriber := func(log *vm.Log, removed bool) {
rpcLog := toRPCLogs(vm.Logs{log}, removed)
if err := subscription.Notify(rpcLog); err != nil {
subscription.Cancel()
} }
} }
// from and to block number are not used since subscriptions don't allow you to travel to "time" subscription, err = notifier.NewSubscription(onUnsubscribe)
var id int
if len(args.Addresses) > 0 {
id, err = s.newLogFilter(-1, -1, args.Addresses, args.Topics, notifySubscriber)
} else {
id, err = s.newLogFilter(-1, -1, nil, args.Topics, notifySubscriber)
}
if err != nil { if err != nil {
subscription.Cancel()
return nil, err return nil, err
} }
s.filterMapMu.Lock() send := func(id FilterID, logs []Log) {
s.filterMapping[externalId] = id rpcLogs := toRPCLogs(logs)
s.filterMapMu.Unlock() for _, l := range rpcLogs {
subscription.Notify(l)
}
}
fid, err := api.m.NewLogFilterWithNoTimeout(crit, send)
if err != nil {
return nil, err
}
api.subscriptionsMu.Lock()
api.subscriptions[subscription.ID()] = fid
api.subscriptionsMu.Unlock()
return subscription, err return subscription, err
} }
// NewFilterArgs represents a request to create a new filter. // UninstallFilter deletes a filter by its id. If successful true is returned,
type NewFilterArgs struct { // otherwise false.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_uninstallfilter
func (api *PublicFilterAPI) UninstallFilter(id FilterID) bool {
return api.m.Uninstall(id) == nil
}
// toRPCLogs is a helper that will return an empty slice of logs when nil is given.
// Otherwise the given logs are returned. This is necessary for the RPC interface.
func toRPCLogs(logs []Log) []Log {
if logs == nil {
return []Log{}
}
return logs
}
// toRPCHashes is a helper that will return an empty hash array case the given hash
// array is nil, otherwise is will return the given hashes. The RPC interfaces defines
// that always an array is returned.
func toRPCHashes(hashes []common.Hash) []common.Hash {
if hashes == nil {
return []common.Hash{}
}
return hashes
}
// FilterCriteria represents a request to create a new filter.
type FilterCriteria struct {
FromBlock rpc.BlockNumber FromBlock rpc.BlockNumber
ToBlock rpc.BlockNumber ToBlock rpc.BlockNumber
Addresses []common.Address Addresses []common.Address
Topics [][]common.Hash Topics [][]common.Hash
} }
// UnmarshalJSON sets *args fields with given data. // UnmarshalJSON sets *args fields with filter criteria JSON encoded in the given data.
func (args *NewFilterArgs) UnmarshalJSON(data []byte) error { func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
type input struct { type input struct {
From *rpc.BlockNumber `json:"fromBlock"` From *rpc.BlockNumber `json:"fromBlock"`
ToBlock *rpc.BlockNumber `json:"toBlock"` ToBlock *rpc.BlockNumber `json:"toBlock"`
@ -405,245 +313,3 @@ func (args *NewFilterArgs) UnmarshalJSON(data []byte) error {
return nil return nil
} }
// NewFilter creates a new filter and returns the filter id. It can be uses to retrieve logs.
func (s *PublicFilterAPI) NewFilter(args NewFilterArgs) (string, error) {
externalId, err := newFilterId()
if err != nil {
return "", err
}
var id int
if len(args.Addresses) > 0 {
id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics, nil)
} else {
id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics, nil)
}
if err != nil {
return "", err
}
s.filterMapMu.Lock()
s.filterMapping[externalId] = id
s.filterMapMu.Unlock()
return externalId, nil
}
// GetLogs returns the logs matching the given argument.
func (s *PublicFilterAPI) GetLogs(args NewFilterArgs) []vmlog {
filter := New(s.chainDb)
filter.SetBeginBlock(args.FromBlock.Int64())
filter.SetEndBlock(args.ToBlock.Int64())
filter.SetAddresses(args.Addresses)
filter.SetTopics(args.Topics)
return toRPCLogs(filter.Find(), false)
}
// UninstallFilter removes the filter with the given filter id.
func (s *PublicFilterAPI) UninstallFilter(filterId string) bool {
s.filterMapMu.Lock()
defer s.filterMapMu.Unlock()
id, ok := s.filterMapping[filterId]
if !ok {
return false
}
defer s.filterManager.Remove(id)
delete(s.filterMapping, filterId)
if _, ok := s.logQueue[id]; ok {
s.logMu.Lock()
defer s.logMu.Unlock()
delete(s.logQueue, id)
return true
}
if _, ok := s.blockQueue[id]; ok {
s.blockMu.Lock()
defer s.blockMu.Unlock()
delete(s.blockQueue, id)
return true
}
if _, ok := s.transactionQueue[id]; ok {
s.transactionMu.Lock()
defer s.transactionMu.Unlock()
delete(s.transactionQueue, id)
return true
}
return false
}
// getFilterType is a helper utility that determine the type of filter for the given filter id.
func (s *PublicFilterAPI) getFilterType(id int) byte {
if _, ok := s.blockQueue[id]; ok {
return blockFilterTy
} else if _, ok := s.transactionQueue[id]; ok {
return transactionFilterTy
} else if _, ok := s.logQueue[id]; ok {
return logFilterTy
}
return unknownFilterTy
}
// blockFilterChanged returns a collection of block hashes for the block filter with the given id.
func (s *PublicFilterAPI) blockFilterChanged(id int) []common.Hash {
s.blockMu.Lock()
defer s.blockMu.Unlock()
if s.blockQueue[id] != nil {
return s.blockQueue[id].get()
}
return nil
}
// transactionFilterChanged returns a collection of transaction hashes for the pending
// transaction filter with the given id.
func (s *PublicFilterAPI) transactionFilterChanged(id int) []common.Hash {
s.blockMu.Lock()
defer s.blockMu.Unlock()
if s.transactionQueue[id] != nil {
return s.transactionQueue[id].get()
}
return nil
}
// logFilterChanged returns a collection of logs for the log filter with the given id.
func (s *PublicFilterAPI) logFilterChanged(id int) []vmlog {
s.logMu.Lock()
defer s.logMu.Unlock()
if s.logQueue[id] != nil {
return s.logQueue[id].get()
}
return nil
}
// GetFilterLogs returns the logs for the filter with the given id.
func (s *PublicFilterAPI) GetFilterLogs(filterId string) []vmlog {
id, ok := s.filterMapping[filterId]
if !ok {
return toRPCLogs(nil, false)
}
if filter := s.filterManager.Get(id); filter != nil {
return toRPCLogs(filter.Find(), false)
}
return toRPCLogs(nil, false)
}
// GetFilterChanges returns the logs for the filter with the given id since last time is was called.
// This can be used for polling.
func (s *PublicFilterAPI) GetFilterChanges(filterId string) interface{} {
s.filterMapMu.Lock()
id, ok := s.filterMapping[filterId]
s.filterMapMu.Unlock()
if !ok { // filter not found
return []interface{}{}
}
switch s.getFilterType(id) {
case blockFilterTy:
return returnHashes(s.blockFilterChanged(id))
case transactionFilterTy:
return returnHashes(s.transactionFilterChanged(id))
case logFilterTy:
return s.logFilterChanged(id)
}
return []interface{}{}
}
type vmlog struct {
*vm.Log
Removed bool `json:"removed"`
}
type logQueue struct {
mu sync.Mutex
logs []vmlog
timeout time.Time
id int
}
func (l *logQueue) add(logs ...vmlog) {
l.mu.Lock()
defer l.mu.Unlock()
l.logs = append(l.logs, logs...)
}
func (l *logQueue) get() []vmlog {
l.mu.Lock()
defer l.mu.Unlock()
l.timeout = time.Now()
tmp := l.logs
l.logs = nil
return tmp
}
type hashQueue struct {
mu sync.Mutex
hashes []common.Hash
timeout time.Time
id int
}
func (l *hashQueue) add(hashes ...common.Hash) {
l.mu.Lock()
defer l.mu.Unlock()
l.hashes = append(l.hashes, hashes...)
}
func (l *hashQueue) get() []common.Hash {
l.mu.Lock()
defer l.mu.Unlock()
l.timeout = time.Now()
tmp := l.hashes
l.hashes = nil
return tmp
}
// newFilterId generates a new random filter identifier that can be exposed to the outer world. By publishing random
// identifiers it is not feasible for DApp's to guess filter id's for other DApp's and uninstall or poll for them
// causing the affected DApp to miss data.
func newFilterId() (string, error) {
var subid [16]byte
n, _ := rand.Read(subid[:])
if n != 16 {
return "", errors.New("Unable to generate filter id")
}
return "0x" + hex.EncodeToString(subid[:]), nil
}
// toRPCLogs is a helper that will convert a vm.Logs array to an structure which
// can hold additional information about the logs such as whether it was deleted.
// Additionally when nil is given it will by default instead create an empty slice
// instead. This is required by the RPC specification.
func toRPCLogs(logs vm.Logs, removed bool) []vmlog {
convertedLogs := make([]vmlog, len(logs))
for i, log := range logs {
convertedLogs[i] = vmlog{Log: log, Removed: removed}
}
return convertedLogs
}
// returnHashes is a helper that will return an empty hash array case the given hash array is nil, otherwise is will
// return the given hashes. The RPC interfaces defines that always an array is returned.
func returnHashes(hashes []common.Hash) []common.Hash {
if hashes == nil {
return []common.Hash{}
}
return hashes
}

19
eth/filters/api_test.go
View file

@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package filters_test package filters
import ( import (
"encoding/json" "encoding/json"
@ -22,7 +22,6 @@ import (
"testing" "testing"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/eth/filters"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
) )
@ -39,7 +38,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
) )
// default values // default values
var test0 filters.NewFilterArgs var test0 FilterCriteria
if err := json.Unmarshal([]byte("{}"), &test0); err != nil { if err := json.Unmarshal([]byte("{}"), &test0); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -57,7 +56,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// from, to block number // from, to block number
var test1 filters.NewFilterArgs var test1 FilterCriteria
vector := fmt.Sprintf(`{"fromBlock":"0x%x","toBlock":"0x%x"}`, fromBlock, toBlock) vector := fmt.Sprintf(`{"fromBlock":"0x%x","toBlock":"0x%x"}`, fromBlock, toBlock)
if err := json.Unmarshal([]byte(vector), &test1); err != nil { if err := json.Unmarshal([]byte(vector), &test1); err != nil {
t.Fatal(err) t.Fatal(err)
@ -70,7 +69,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// single address // single address
var test2 filters.NewFilterArgs var test2 FilterCriteria
vector = fmt.Sprintf(`{"address": "%s"}`, address0.Hex()) vector = fmt.Sprintf(`{"address": "%s"}`, address0.Hex())
if err := json.Unmarshal([]byte(vector), &test2); err != nil { if err := json.Unmarshal([]byte(vector), &test2); err != nil {
t.Fatal(err) t.Fatal(err)
@ -83,7 +82,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// multiple address // multiple address
var test3 filters.NewFilterArgs var test3 FilterCriteria
vector = fmt.Sprintf(`{"address": ["%s", "%s"]}`, address0.Hex(), address1.Hex()) vector = fmt.Sprintf(`{"address": ["%s", "%s"]}`, address0.Hex(), address1.Hex())
if err := json.Unmarshal([]byte(vector), &test3); err != nil { if err := json.Unmarshal([]byte(vector), &test3); err != nil {
t.Fatal(err) t.Fatal(err)
@ -99,7 +98,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// single topic // single topic
var test4 filters.NewFilterArgs var test4 FilterCriteria
vector = fmt.Sprintf(`{"topics": ["%s"]}`, topic0.Hex()) vector = fmt.Sprintf(`{"topics": ["%s"]}`, topic0.Hex())
if err := json.Unmarshal([]byte(vector), &test4); err != nil { if err := json.Unmarshal([]byte(vector), &test4); err != nil {
t.Fatal(err) t.Fatal(err)
@ -115,7 +114,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// test multiple "AND" topics // test multiple "AND" topics
var test5 filters.NewFilterArgs var test5 FilterCriteria
vector = fmt.Sprintf(`{"topics": ["%s", "%s"]}`, topic0.Hex(), topic1.Hex()) vector = fmt.Sprintf(`{"topics": ["%s", "%s"]}`, topic0.Hex(), topic1.Hex())
if err := json.Unmarshal([]byte(vector), &test5); err != nil { if err := json.Unmarshal([]byte(vector), &test5); err != nil {
t.Fatal(err) t.Fatal(err)
@ -137,7 +136,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// test optional topic // test optional topic
var test6 filters.NewFilterArgs var test6 FilterCriteria
vector = fmt.Sprintf(`{"topics": ["%s", null, "%s"]}`, topic0.Hex(), topic2.Hex()) vector = fmt.Sprintf(`{"topics": ["%s", null, "%s"]}`, topic0.Hex(), topic2.Hex())
if err := json.Unmarshal([]byte(vector), &test6); err != nil { if err := json.Unmarshal([]byte(vector), &test6); err != nil {
t.Fatal(err) t.Fatal(err)
@ -165,7 +164,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
} }
// test OR topics // test OR topics
var test7 filters.NewFilterArgs var test7 FilterCriteria
vector = fmt.Sprintf(`{"topics": [["%s", "%s"], null, ["%s", null]]}`, topic0.Hex(), topic1.Hex(), topic2.Hex()) vector = fmt.Sprintf(`{"topics": [["%s", "%s"], null, ["%s", null]]}`, topic0.Hex(), topic1.Hex(), topic2.Hex())
if err := json.Unmarshal([]byte(vector), &test7); err != nil { if err := json.Unmarshal([]byte(vector), &test7); err != nil {
t.Fatal(err) t.Fatal(err)

109
eth/filters/filter.go
View file

@ -23,15 +23,10 @@ import (
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
) )
type AccountChange struct { // Filter can be used to retrieve and filter logs
Address, StateAddress []byte
}
// Filtering interface
type Filter struct { type Filter struct {
created time.Time created time.Time
@ -39,67 +34,67 @@ type Filter struct {
begin, end int64 begin, end int64
addresses []common.Address addresses []common.Address
topics [][]common.Hash topics [][]common.Hash
BlockCallback func(*types.Block, vm.Logs)
TransactionCallback func(*types.Transaction)
LogCallback func(*vm.Log, bool)
} }
// Create a new filter which uses a bloom filter on blocks to figure out whether a particular block // New creates a new filter which uses a bloom filter on blocks to figure out whether
// is interesting or not. // a particular block is interesting or not.
func New(db ethdb.Database) *Filter { func New(db ethdb.Database) *Filter {
return &Filter{db: db} return &Filter{db: db}
} }
// Set the earliest and latest block for filtering. // SetBeginBlock sets the earliest block for filtering.
// -1 = latest block (i.e., the current block) // -1 = latest block (i.e., the current block)
// hash = particular hash from-to // hash = particular hash from-to
func (self *Filter) SetBeginBlock(begin int64) { func (f *Filter) SetBeginBlock(begin int64) {
self.begin = begin f.begin = begin
} }
func (self *Filter) SetEndBlock(end int64) { // SetEndBlock sets the latest block for filtering.
self.end = end func (f *Filter) SetEndBlock(end int64) {
f.end = end
} }
func (self *Filter) SetAddresses(addr []common.Address) { // SetAddresses matches only logs that are generated from addresses that are included
self.addresses = addr // in the given addresses.
func (f *Filter) SetAddresses(addr []common.Address) {
f.addresses = addr
} }
func (self *Filter) SetTopics(topics [][]common.Hash) { // SetTopics matches only logs that have topics matching the given topics.
self.topics = topics func (f *Filter) SetTopics(topics [][]common.Hash) {
f.topics = topics
} }
// Run filters logs with the current parameters set // Find logs with the current parameters set.
func (self *Filter) Find() vm.Logs { func (f *Filter) Find() []Log {
latestHash := core.GetHeadBlockHash(self.db) latestHash := core.GetHeadBlockHash(f.db)
latestBlock := core.GetBlock(self.db, latestHash, core.GetBlockNumber(self.db, latestHash)) latestBlock := core.GetBlock(f.db, latestHash, core.GetBlockNumber(f.db, latestHash))
var beginBlockNo uint64 = uint64(self.begin) beginBlockNo := uint64(f.begin)
if self.begin == -1 { if f.begin == -1 {
beginBlockNo = latestBlock.NumberU64() beginBlockNo = latestBlock.NumberU64()
} }
var endBlockNo uint64 = uint64(self.end) endBlockNo := uint64(f.end)
if self.end == -1 { if f.end == -1 {
endBlockNo = latestBlock.NumberU64() endBlockNo = latestBlock.NumberU64()
} }
// if no addresses are present we can't make use of fast search which // if no addresses are present we can't make use of fast search which
// uses the mipmap bloom filters to check for fast inclusion and uses // uses the mipmap bloom filters to check for fast inclusion and uses
// higher range probability in order to ensure at least a false positive // higher range probability in order to ensure at least a false positive
if len(self.addresses) == 0 { if len(f.addresses) == 0 {
return self.getLogs(beginBlockNo, endBlockNo) return f.getLogs(beginBlockNo, endBlockNo)
} }
return self.mipFind(beginBlockNo, endBlockNo, 0) return f.mipFind(beginBlockNo, endBlockNo, 0)
} }
func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) { func (f *Filter) mipFind(start, end uint64, depth int) (logs []Log) {
level := core.MIPMapLevels[depth] level := core.MIPMapLevels[depth]
// normalise numerator so we can work in level specific batches and // normalise numerator so we can work in level specific batches and
// work with the proper range checks // work with the proper range checks
for num := start / level * level; num <= end; num += level { for num := start / level * level; num <= end; num += level {
// find addresses in bloom filters // find addresses in bloom filters
bloom := core.GetMipmapBloom(self.db, num, level) bloom := core.GetMipmapBloom(f.db, num, level)
for _, addr := range self.addresses { for _, addr := range f.addresses {
if bloom.TestBytes(addr[:]) { if bloom.TestBytes(addr[:]) {
// range check normalised values and make sure that // range check normalised values and make sure that
// we're resolving the correct range instead of the // we're resolving the correct range instead of the
@ -107,9 +102,9 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) {
start := uint64(math.Max(float64(num), float64(start))) start := uint64(math.Max(float64(num), float64(start)))
end := uint64(math.Min(float64(num+level-1), float64(end))) end := uint64(math.Min(float64(num+level-1), float64(end)))
if depth+1 == len(core.MIPMapLevels) { if depth+1 == len(core.MIPMapLevels) {
logs = append(logs, self.getLogs(start, end)...) logs = append(logs, f.getLogs(start, end)...)
} else { } else {
logs = append(logs, self.mipFind(start, end, depth+1)...) logs = append(logs, f.mipFind(start, end, depth+1)...)
} }
// break so we don't check the same range for each // break so we don't check the same range for each
// possible address. Checks on multiple addresses // possible address. Checks on multiple addresses
@ -122,29 +117,33 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) {
return logs return logs
} }
func (self *Filter) getLogs(start, end uint64) (logs vm.Logs) { func (f *Filter) getLogs(start, end uint64) (logs []Log) {
var block *types.Block var block *types.Block
for i := start; i <= end; i++ { for i := start; i <= end; i++ {
hash := core.GetCanonicalHash(self.db, i) hash := core.GetCanonicalHash(f.db, i)
if hash != (common.Hash{}) { if hash != (common.Hash{}) {
block = core.GetBlock(self.db, hash, i) block = core.GetBlock(f.db, hash, i)
} else { // block not found } else { // block not found
return logs return logs
} }
// Use bloom filtering to see if this block is interesting given the // Use bloom filtering to see if this block is interesting given the
// current parameters // current parameters
if self.bloomFilter(block) { if f.bloomFilter(block) {
// Get the logs of the block // Get the logs of the block
var ( var (
receipts = core.GetBlockReceipts(self.db, block.Hash(), i) receipts = core.GetBlockReceipts(f.db, block.Hash(), i)
unfiltered vm.Logs unfiltered []Log
) )
for _, receipt := range receipts { for _, receipt := range receipts {
unfiltered = append(unfiltered, receipt.Logs...) rl := make([]Log, len(receipt.Logs))
for i, l := range receipt.Logs {
rl[i] = Log{l, false}
}
unfiltered = append(unfiltered, rl...)
} }
logs = append(logs, self.FilterLogs(unfiltered)...) logs = append(logs, filterLogs(unfiltered, f.addresses, f.topics)...)
} }
} }
@ -161,26 +160,26 @@ func includes(addresses []common.Address, a common.Address) bool {
return false return false
} }
func (self *Filter) FilterLogs(logs vm.Logs) vm.Logs { func filterLogs(logs []Log, addresses []common.Address, topics [][]common.Hash) []Log {
var ret vm.Logs var ret []Log
// Filter the logs for interesting stuff // Filter the logs for interesting stuff
Logs: Logs:
for _, log := range logs { for _, log := range logs {
if len(self.addresses) > 0 && !includes(self.addresses, log.Address) { if len(addresses) > 0 && !includes(addresses, log.Address) {
continue continue
} }
logTopics := make([]common.Hash, len(self.topics)) logTopics := make([]common.Hash, len(topics))
copy(logTopics, log.Topics) copy(logTopics, log.Topics)
// If the to filtered topics is greater than the amount of topics in // If the to filtered topics is greater than the amount of topics in
// logs, skip. // logs, skip.
if len(self.topics) > len(log.Topics) { if len(topics) > len(log.Topics) {
continue Logs continue Logs
} }
for i, topics := range self.topics { for i, topics := range topics {
var match bool var match bool
for _, topic := range topics { for _, topic := range topics {
// common.Hash{} is a match all (wildcard) // common.Hash{} is a match all (wildcard)
@ -202,10 +201,10 @@ Logs:
return ret return ret
} }
func (self *Filter) bloomFilter(block *types.Block) bool { func (f *Filter) bloomFilter(block *types.Block) bool {
if len(self.addresses) > 0 { if len(f.addresses) > 0 {
var included bool var included bool
for _, addr := range self.addresses { for _, addr := range f.addresses {
if types.BloomLookup(block.Bloom(), addr) { if types.BloomLookup(block.Bloom(), addr) {
included = true included = true
break break
@ -217,7 +216,7 @@ func (self *Filter) bloomFilter(block *types.Block) bool {
} }
} }
for _, sub := range self.topics { for _, sub := range f.topics {
var included bool var included bool
for _, topic := range sub { for _, topic := range sub {
if (topic == common.Hash{}) || types.BloomLookup(block.Bloom(), topic) { if (topic == common.Hash{}) || types.BloomLookup(block.Bloom(), topic) {

642
eth/filters/filter_system.go
View file

@ -14,172 +14,576 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// package filters implements an ethereum filtering system for block, // Package filters implements an ethereum filtering system for block,
// transactions and log events. // transactions and log events.
package filters package filters
import ( import (
"bufio"
crand "crypto/rand"
"encoding/binary"
"encoding/hex"
"encoding/json"
"errors"
"fmt" "fmt"
"math/rand"
"sync" "sync"
"time" "time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
) )
// FilterType determines the type of filter and is used to put the filter in to
// the correct bucket when added.
type FilterType byte
const ( const (
ChainFilter FilterType = iota // new block events filter // UnknownFilter indicates an unkown filter type
PendingTxFilter // pending transaction filter UnknownFilter Type = iota
LogFilter // new or removed log filter // BlockFilter queries for new blocks
PendingLogFilter // pending log filter BlockFilter
// PendingTxFilter queries pending transactions
PendingTxFilter
// LogFilter queries for new or removed (chain reorg) logs
LogFilter
// PendingLogFilter queries for logs for the pending block
PendingLogFilter
maxPendingHashes = 10240 // max buffer size of block/tx hashes before an filter is considered inactive and disabled
maxPendingLogs = 10240 // max buffer size of logs before an filter is considered inactive and disabled
) )
// FilterSystem manages filters that filter specific events such as var (
// block, transaction and log events. The Filtering system can be used to listen errFilterNotFound = errors.New("filter not found")
// for specific LOG events fired by the EVM (Ethereum Virtual Machine). errCouldNotGenerateFilterID = errors.New("unable to generate filter id")
type FilterSystem struct { errInvalidFilterID = errors.New("invalid filter id")
filterMu sync.RWMutex errUnableToUninstallFilter = errors.New("unable to uninstall filter, retry later")
filterId int
chainFilters map[int]*Filter filterIDGenMu sync.Mutex
pendingTxFilters map[int]*Filter filterIDGen = filterIDGenerator()
logFilters map[int]*Filter )
pendingLogFilters map[int]*Filter
// generic is an ugly hack for Get type logsCallback func(FilterID, []Log)
generic map[int]*Filter
sub event.Subscription // Type determines the kind of filter and is used to put the filter in to
// the correct bucket when added.
type Type byte
// FilterID determines the type for a filter identifier.
type FilterID [16]byte
// MarshalJSON serializes a FilterID into its JSON representation.
func (f FilterID) MarshalJSON() ([]byte, error) {
return json.Marshal(fmt.Sprintf("0x%x", f))
} }
// NewFilterSystem returns a newly allocated filter manager // filterIDGenerator helper utility that generates a (pseudo) random sequence of bytes
func NewFilterSystem(mux *event.TypeMux) *FilterSystem { // that are used to generate filter identifiers.
fs := &FilterSystem{ func filterIDGenerator() *rand.Rand {
chainFilters: make(map[int]*Filter), if seed, err := binary.ReadVarint(bufio.NewReader(crand.Reader)); err == nil {
pendingTxFilters: make(map[int]*Filter), return rand.New(rand.NewSource(seed))
logFilters: make(map[int]*Filter),
pendingLogFilters: make(map[int]*Filter),
generic: make(map[int]*Filter),
} }
fs.sub = mux.Subscribe( return rand.New(rand.NewSource(int64(time.Now().Nanosecond())))
}
// newID generates a filter identifier.
func newID() FilterID {
filterIDGenMu.Lock()
defer filterIDGenMu.Unlock()
var id FilterID
for i := 0; i < len(id); i += 7 {
val := filterIDGen.Int63()
for j := 0; i+j < len(id) && j < 7; j++ {
id[i+j] = byte(val)
val >>= 8
}
}
return id
}
// UnmarshalJSON parses a FilterID from its JSON representation.
func (f *FilterID) UnmarshalJSON(data []byte) error {
// `"0x...."`
if len(data) != 36 || data[0] != '"' || data[35] != '"' || data[1] != '0' || (data[2] != 'x' && data[2] != 'X') {
return errInvalidFilterID
}
_, err := hex.Decode(f[:], data[3:35])
return err
}
// Log is a helper that can hold additional information about vm.Log
// necessary for the RPC interface.
type Log struct {
*vm.Log
Removed bool `json:"removed"`
}
type filter struct {
ID FilterID
typ Type
created time.Time
lastUsed time.Time
canTimeout bool
hashes chan common.Hash // results in case filter type returns hashes
logsCrit FilterCriteria
lc logsCallback
logs chan []Log // results in case filter type returns logs
}
// Manager listens for new events and offers a filter system to queury for
// events that match a set of criteria.
//
// Note, this system cannot be used to query past logs, it will only receive
// and handle events that are posted by the global event mux. Use a raw Filter
// to query for logs that are already stored.
type Manager struct {
sub event.Subscription
install chan *filter // install filter for event notification
uninstall chan *filter // remove filter for event notification
allMu sync.RWMutex
all map[FilterID]*filter // all installed filters
}
// NewManager creates a new manager that listens for event on the given mux,
// parses and filters them. It uses the all map to retrieve filter changes. The
// work loop holds its own index that is used to forward events to filters.
//
// The returned manager has a loop that needs to be stopped with the Stop function
// or by stopping the given mux.
func NewManager(mux *event.TypeMux) *Manager {
sub := mux.Subscribe(
core.PendingLogsEvent{}, core.PendingLogsEvent{},
core.RemovedLogsEvent{}, core.RemovedLogsEvent{},
core.ChainEvent{}, core.ChainEvent{},
core.TxPreEvent{}, core.TxPreEvent{},
vm.Logs(nil), vm.Logs(nil),
) )
go fs.filterLoop()
return fs
}
// Stop quits the filter loop required for polling events m := &Manager{
func (fs *FilterSystem) Stop() { sub: sub,
fs.sub.Unsubscribe() install: make(chan *filter),
} uninstall: make(chan *filter, 1024),
all: make(map[FilterID]*filter),
// Add adds a filter to the filter manager
func (fs *FilterSystem) Add(filter *Filter, filterType FilterType) (int, error) {
fs.filterMu.Lock()
defer fs.filterMu.Unlock()
id := fs.filterId
filter.created = time.Now()
switch filterType {
case ChainFilter:
fs.chainFilters[id] = filter
case PendingTxFilter:
fs.pendingTxFilters[id] = filter
case LogFilter:
fs.logFilters[id] = filter
case PendingLogFilter:
fs.pendingLogFilters[id] = filter
default:
return 0, fmt.Errorf("unknown filter type %v", filterType)
} }
fs.generic[id] = filter
fs.filterId++ go m.run()
return m
}
// Uninstall filter. If the given filter could not be found an error is returned.
func (m *Manager) Uninstall(id FilterID) error {
m.allMu.Lock()
if f, found := m.all[id]; found {
delete(m.all, id)
m.allMu.Unlock()
select {
case m.uninstall <- f:
return nil
default:
// can only happen when there are too many pending uninstall requests
return errUnableToUninstallFilter
}
}
m.allMu.Unlock()
return errFilterNotFound
}
// FilterType returns the filter type for the given id.
// If the filter could be found UnknownFilter is returned.
func (m *Manager) FilterType(id FilterID) Type {
m.allMu.RLock()
defer m.allMu.RUnlock()
if f, found := m.all[id]; found {
return f.typ
}
return UnknownFilter
}
// NewBlockFilter returns a filter identifier that can be used to get the hashes
// for new blocks. The given callback is optional. If nil is given block hashes
// are queued until fetched with GetBlockFilterChanges.
func (m *Manager) NewBlockFilter() (FilterID, error) {
id := newID()
f := &filter{
ID: id,
created: time.Now(),
lastUsed: time.Now(),
canTimeout: true,
typ: BlockFilter,
hashes: make(chan common.Hash, maxPendingHashes),
logs: make(chan []Log, maxPendingLogs),
}
m.allMu.Lock()
m.all[id] = f
m.allMu.Unlock()
m.install <- f
return id, nil return id, nil
} }
// Remove removes a filter by filter id // GetBlockFilterChanges returns the data for the filter with the given id since
func (fs *FilterSystem) Remove(id int) { // last time it was called.
fs.filterMu.Lock() func (m *Manager) GetBlockFilterChanges(id FilterID) ([]common.Hash, error) {
defer fs.filterMu.Unlock() m.allMu.RLock()
if f, found := m.all[id]; found && f.typ == BlockFilter {
m.allMu.RUnlock()
delete(fs.chainFilters, id) // retieve hashes
delete(fs.pendingTxFilters, id) f.lastUsed = time.Now()
delete(fs.logFilters, id) hashes := make([]common.Hash, 0, len(f.hashes)) // prevent (most) allocs
delete(fs.pendingLogFilters, id) for {
delete(fs.generic, id) select {
case h := <-f.hashes:
hashes = append(hashes, h)
default:
return hashes, nil
}
}
}
m.allMu.RUnlock()
return nil, errFilterNotFound
} }
func (fs *FilterSystem) Get(id int) *Filter { // NewLogFilterWithNoTimeout returns a filter identifier that can be used to fetch
fs.filterMu.RLock() // logs matching the given criteria. The created filter will not timeout and the
defer fs.filterMu.RUnlock() // callee is expected to manually uninstall the filter.
func (m *Manager) NewLogFilterWithNoTimeout(crit FilterCriteria, cb logsCallback) (FilterID, error) {
id := newID()
return fs.generic[id] f := &filter{
ID: id,
created: time.Now(),
lastUsed: time.Now(),
canTimeout: false,
typ: LogFilter,
hashes: make(chan common.Hash, maxPendingHashes),
logs: make(chan []Log, maxPendingLogs),
lc: cb,
logsCrit: crit,
}
m.allMu.Lock()
m.all[id] = f
m.allMu.Unlock()
m.install <- f
return id, nil
} }
// filterLoop waits for specific events from ethereum and fires their handlers // NewLogFilter returns a filter identifier that can be used to fetch logs matching
// when the filter matches the requirements. // the given criteria.
func (fs *FilterSystem) filterLoop() { func (m *Manager) NewLogFilter(crit FilterCriteria, cb logsCallback) (FilterID, error) {
for event := range fs.sub.Chan() { id := newID()
switch ev := event.Data.(type) {
case core.ChainEvent:
fs.filterMu.RLock()
for _, filter := range fs.chainFilters {
if filter.BlockCallback != nil && !filter.created.After(event.Time) {
filter.BlockCallback(ev.Block, ev.Logs)
}
}
fs.filterMu.RUnlock()
case core.TxPreEvent:
fs.filterMu.RLock()
for _, filter := range fs.pendingTxFilters {
if filter.TransactionCallback != nil && !filter.created.After(event.Time) {
filter.TransactionCallback(ev.Tx)
}
}
fs.filterMu.RUnlock()
case vm.Logs: f := &filter{
fs.filterMu.RLock() ID: id,
for _, filter := range fs.logFilters { created: time.Now(),
if filter.LogCallback != nil && !filter.created.After(event.Time) { lastUsed: time.Now(),
for _, log := range filter.FilterLogs(ev) { canTimeout: true,
filter.LogCallback(log, false) typ: LogFilter,
} hashes: make(chan common.Hash, maxPendingHashes),
logs: make(chan []Log, maxPendingLogs),
lc: cb,
logsCrit: crit,
}
m.allMu.Lock()
m.all[id] = f
m.allMu.Unlock()
m.install <- f
return id, nil
}
// GetLogFilterCriteria returns the filtering criteria for a filter, or an error in
// case the filter could not be found.
func (m *Manager) GetLogFilterCriteria(id FilterID) (FilterCriteria, error) {
m.allMu.RLock()
defer m.allMu.RUnlock()
if f, found := m.all[id]; found {
return f.logsCrit, nil
}
return FilterCriteria{}, errFilterNotFound
}
// NewPendingLogFilter creates a filter that returns new pending logs that match the given criteria.
func (m *Manager) NewPendingLogFilter(crit FilterCriteria, cb logsCallback) (FilterID, error) {
id := newID()
f := &filter{
ID: id,
created: time.Now(),
lastUsed: time.Now(),
canTimeout: true,
typ: PendingLogFilter,
hashes: make(chan common.Hash, maxPendingHashes),
logs: make(chan []Log, maxPendingLogs),
lc: cb,
logsCrit: crit,
}
m.allMu.Lock()
m.all[id] = f
m.allMu.Unlock()
m.install <- f
return id, nil
}
// GetPendingLogFilterChanges returns logs for the pending block.
func (m *Manager) GetPendingLogFilterChanges(id FilterID) ([]Log, error) {
m.allMu.RLock()
defer m.allMu.RUnlock()
if f, found := m.all[id]; found && f.typ == PendingLogFilter {
f.lastUsed = time.Now()
allLogs := make([]Log, 0, len(f.logs)) // prevent (most) allocs
for {
select {
case logs := <-f.logs:
allLogs = append(allLogs, logs...)
default: // available logs read
return allLogs, nil
}
}
}
return nil, errFilterNotFound
}
// GetLogFilterChanges returns all logs matching the criteria for the filter with the given filter id.
func (m *Manager) GetLogFilterChanges(id FilterID) ([]Log, error) {
m.allMu.RLock()
defer m.allMu.RUnlock()
if f, found := m.all[id]; found && f.typ == LogFilter {
f.lastUsed = time.Now()
allLogs := make([]Log, 0, len(f.logs)) // prevent (most) allocs for the append
for {
select {
case logs := <-f.logs:
allLogs = append(allLogs, logs...)
default: // available logs read
return allLogs, nil
}
}
}
return nil, errFilterNotFound
}
// NewPendingTransactionFilter creates a filter that retrieves pending transactions.
func (m *Manager) NewPendingTransactionFilter() (FilterID, error) {
id := newID()
f := &filter{
ID: id,
created: time.Now(),
lastUsed: time.Now(),
canTimeout: true,
typ: PendingTxFilter,
hashes: make(chan common.Hash, maxPendingHashes),
logs: make(chan []Log, maxPendingLogs),
}
m.allMu.Lock()
m.all[id] = f
m.allMu.Unlock()
m.install <- f
return id, nil
}
// GetPendingTxFilterChanges returns hashes for pending transactions which are added since the last poll.
func (m *Manager) GetPendingTxFilterChanges(id FilterID) ([]common.Hash, error) {
m.allMu.RLock()
defer m.allMu.RUnlock()
if f, found := m.all[id]; found && f.typ == PendingTxFilter {
f.lastUsed = time.Now()
hashes := make([]common.Hash, 0, len(f.hashes)) // prevent (most) allocs
for {
select {
case hash := <-f.hashes:
hashes = append(hashes, hash)
default: // read available tx hashes
return hashes, nil
}
}
}
return nil, errFilterNotFound
}
type filterIndex map[Type]map[FilterID]*filter
// process an event and forward to filters that match the criteria.
func (m *Manager) process(filters filterIndex, ev *event.Event) {
var inactive []*filter
logHandler := func(f *filter, logs []Log) {
if f.lc != nil && len(logs) > 0 {
f.lc(f.ID, logs)
} else if len(logs) > 0 {
select {
case f.logs <- logs:
return
default: // data queue full, disable filter
inactive = append(inactive, f)
}
}
}
switch e := ev.Data.(type) {
case core.ChainEvent:
for _, f := range filters[BlockFilter] {
if ev.Time.After(f.created) {
select {
case f.hashes <- e.Hash:
continue
default:
// data queue full, disable filter
inactive = append(inactive, f)
} }
} }
fs.filterMu.RUnlock() }
case core.RemovedLogsEvent: case core.TxPreEvent:
fs.filterMu.RLock() for _, f := range filters[PendingTxFilter] {
for _, filter := range fs.logFilters { if ev.Time.After(f.created) {
if filter.LogCallback != nil && !filter.created.After(event.Time) { select {
for _, removedLog := range filter.FilterLogs(ev.Logs) { case f.hashes <- e.Tx.Hash():
filter.LogCallback(removedLog, true) continue
} default:
// data queue full, disable filter
inactive = append(inactive, f)
} }
} }
fs.filterMu.RUnlock() }
case core.PendingLogsEvent: case vm.Logs:
fs.filterMu.RLock() for _, f := range filters[LogFilter] {
for _, filter := range fs.pendingLogFilters { if ev.Time.After(f.created) {
if filter.LogCallback != nil && !filter.created.After(event.Time) { matchedLogs := filterLogs(convertLogs(e, false), f.logsCrit.Addresses, f.logsCrit.Topics)
for _, pendingLog := range ev.Logs { logHandler(f, matchedLogs)
filter.LogCallback(pendingLog, false)
}
}
} }
fs.filterMu.RUnlock() }
case core.RemovedLogsEvent:
for _, f := range filters[LogFilter] {
if ev.Time.After(f.created) {
matchedLogs := filterLogs(convertLogs(e.Logs, true), f.logsCrit.Addresses, f.logsCrit.Topics)
logHandler(f, matchedLogs)
}
}
case core.PendingLogsEvent:
for _, f := range filters[PendingLogFilter] {
if ev.Time.After(f.created) {
matchedLogs := filterLogs(convertLogs(e.Logs, false), f.logsCrit.Addresses, f.logsCrit.Topics)
logHandler(f, matchedLogs)
}
}
}
m.allMu.Lock()
for _, f := range inactive {
delete(m.all, f.ID)
glog.Warningf("filter 0x%x uninstalled, queue full\n", f.ID)
}
m.allMu.Unlock()
// remove filter for event listening, this must be run in a seperate go routine
// since timeout is called from the work loop and sending uninstall requests to the
// work loop from "itself" may deadlock when the uninstall channel is full.
go func() {
for _, f := range inactive {
m.uninstall <- f
}
}()
}
// timeout uninstalls all filters that have not been used in the last 5 minutes.
func (m *Manager) timeout() {
deadline := time.Now().Add(-5 * time.Minute)
var inactive []*filter
m.allMu.Lock()
for _, f := range m.all {
if f.lastUsed.Before(deadline) && f.canTimeout {
delete(m.all, f.ID) // filter cannot be used from the external
inactive = append(inactive, f)
}
}
m.allMu.Unlock()
// remove filter for event listening, this must be run in a seperate go routine
// since timeout is called from the work loop and sending uninstall requests to the
// work loop from "itself" may deadlock when the uninstall channel is full.
go func() {
for _, f := range inactive {
m.uninstall <- f // delete for the internal work loop
}
}()
}
// run is the manager work loop.
// It will receive events and forwards them to installed filters.
// Inactive filters will be uninstalled.
func (m *Manager) run() {
index := make(filterIndex)
timeout := time.NewTicker(30 * time.Second)
for {
select {
case f := <-m.install:
// lazy load
if _, found := index[f.typ]; !found {
index[f.typ] = make(map[FilterID]*filter)
}
index[f.typ][f.ID] = f
case f := <-m.uninstall:
close(f.hashes)
close(f.logs)
delete(index[f.typ], f.ID)
case ev, ok := <-m.sub.Chan():
if !ok {
glog.V(logger.Debug).Infoln("filter manager stopped")
return
}
m.process(index, ev)
case <-timeout.C:
m.timeout()
} }
} }
} }
// Stop the filter system.
func (m *Manager) Stop() {
m.sub.Unsubscribe() // end worker loop
}
// convertLogs is a helper utility that converts vm.Logs to []filter.Log.
func convertLogs(in vm.Logs, removed bool) []Log {
logs := make([]Log, len(in))
for i, l := range in {
logs[i] = Log{l, false}
}
return logs
}

501
eth/filters/filter_system_test.go
View file

@ -17,101 +17,442 @@
package filters package filters
import ( import (
"encoding/json"
"math/big"
"reflect"
"testing" "testing"
"time" "time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/rpc"
) )
func TestCallbacks(t *testing.T) { var (
var ( mux = new(event.TypeMux)
mux event.TypeMux filterManager = NewManager(mux)
fs = NewFilterSystem(&mux) )
blockDone = make(chan struct{})
txDone = make(chan struct{})
logDone = make(chan struct{})
removedLogDone = make(chan struct{})
pendingLogDone = make(chan struct{})
)
blockFilter := &Filter{ // TestFilterIdSerialization tests if FilterIs is correct serialized and parsed.
BlockCallback: func(*types.Block, vm.Logs) { func TestFilterIdSerialization(t *testing.T) {
close(blockDone) t.Parallel()
},
} id := newID()
txFilter := &Filter{
TransactionCallback: func(*types.Transaction) { serialized, err := json.Marshal(id)
close(txDone) if err != nil {
}, t.Fatal(err)
}
logFilter := &Filter{
LogCallback: func(l *vm.Log, oob bool) {
if !oob {
close(logDone)
}
},
}
removedLogFilter := &Filter{
LogCallback: func(l *vm.Log, oob bool) {
if oob {
close(removedLogDone)
}
},
}
pendingLogFilter := &Filter{
LogCallback: func(*vm.Log, bool) {
close(pendingLogDone)
},
} }
fs.Add(blockFilter, ChainFilter) if len(serialized) != 36 {
fs.Add(txFilter, PendingTxFilter) t.Fatalf("invalid filter id length (%s), want %d, got %d", serialized, 36, len(serialized))
fs.Add(logFilter, LogFilter)
fs.Add(removedLogFilter, LogFilter)
fs.Add(pendingLogFilter, PendingLogFilter)
mux.Post(core.ChainEvent{})
mux.Post(core.TxPreEvent{})
mux.Post(vm.Logs{&vm.Log{}})
mux.Post(core.RemovedLogsEvent{Logs: vm.Logs{&vm.Log{}}})
mux.Post(core.PendingLogsEvent{Logs: vm.Logs{&vm.Log{}}})
const dura = 5 * time.Second
failTimer := time.NewTimer(dura)
select {
case <-blockDone:
case <-failTimer.C:
t.Error("block filter failed to trigger (timeout)")
} }
failTimer.Reset(dura) var filterID FilterID
select { if err := json.Unmarshal(serialized, &filterID); err != nil {
case <-txDone: t.Fatal(err)
case <-failTimer.C:
t.Error("transaction filter failed to trigger (timeout)")
} }
failTimer.Reset(dura) if filterID != id {
select { t.Errorf("invalid filter id, want %x, got %x", id, filterID)
case <-logDone: }
case <-failTimer.C: }
t.Error("log filter failed to trigger (timeout)")
} // TestBlockFilter tests if a block filter callback is called when core.ChainEvent are posted.
// It creates multiple filters:
failTimer.Reset(dura) // - one at the start and should receive all posted chain events and a second (blockHashes)
select { // - one that is created after a cutoff moment and uninstalled after a second cutoff moment (blockHashes[cutoff1:cutoff2])
case <-removedLogDone: // - one that is created after the second cutoff moment (blockHashes[cutoff2:])
case <-failTimer.C: func TestBlockFilter(t *testing.T) {
t.Error("removed log filter failed to trigger (timeout)") t.Parallel()
}
var (
failTimer.Reset(dura) blockHashes = []common.Hash{
select { common.HexToHash("0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"),
case <-pendingLogDone: common.HexToHash("0x88e96d4537bea4d9c05d12549907b32561d3bf31f45aae734cdc119f13406cb6"),
case <-failTimer.C: common.HexToHash("0xb495a1d7e6663152ae92708da4843337b958146015a2802f4193a410044698c9"),
t.Error("pending log filter failed to trigger (timeout)") common.HexToHash("0x3d6122660cc824376f11ee842f83addc3525e2dd6756b9bcf0affa6aa88cf741"),
common.HexToHash("0x23adf5a3be0f5235b36941bcb29b62504278ec5b9cdfa277b992ba4a7a3cd3a2"),
common.HexToHash("0xf37c632d361e0a93f08ba29b1a2c708d9caa3ee19d1ee8d2a02612bffe49f0a9"),
common.HexToHash("0x1f1aed8e3694a067496c248e61879cda99b0709a1dfbacd0b693750df06b326e"),
common.HexToHash("0xe0c7c0b46e116b874354dce6f64b8581bd239186b03f30a978e3dc38656f723a"),
common.HexToHash("0x2ce94342df186bab4165c268c43ab982d360c9474f429fec5565adfc5d1f258b"),
common.HexToHash("0x997e47bf4cac509c627753c06385ac866641ec6f883734ff7944411000dc576e"),
}
cutoff1 = 3
cutoff2 = cutoff1 + 2
receivedHashes1 []common.Hash
receivedHashes2 []common.Hash
receivedHashes3 []common.Hash
)
// should receive all block hashes
fid1, err := filterManager.NewBlockFilter()
if err != nil {
t.Fatal(err)
}
for _, h := range blockHashes[:cutoff1] {
ev := core.ChainEvent{Hash: h}
mux.Post(ev)
}
// fid1 receives blockHashes[:]
for {
hashes, err := filterManager.GetBlockFilterChanges(fid1)
if err != nil {
t.Fatalf("unable to fetch block filter changes: %v", err)
}
receivedHashes1 = append(receivedHashes1, hashes...)
if len(receivedHashes1) >= cutoff1 {
break
}
time.Sleep(100 * time.Millisecond)
}
// fid2 receives blockHashes[cutoff1:cutoff2]
fid2, err := filterManager.NewBlockFilter()
if err != nil {
t.Fatal(err)
}
for _, h := range blockHashes[cutoff1:cutoff2] {
ev := core.ChainEvent{Hash: h}
mux.Post(ev)
}
for len(receivedHashes1) < cutoff2 {
hashes, err := filterManager.GetBlockFilterChanges(fid1)
if err != nil {
t.Fatalf("unable to fetch block filter changes: %v", err)
}
receivedHashes1 = append(receivedHashes1, hashes...)
}
receivedHashes2, err = filterManager.GetBlockFilterChanges(fid2)
if err != nil {
t.Fatalf("unable to fetch block filter changes: %v", err)
}
// delete second filter to test we don't get any more events anymore for the filter
if err := filterManager.Uninstall(fid2); err != nil {
t.Fatalf("uninstall filter failed with %v", err)
}
if _, err = filterManager.GetBlockFilterChanges(fid2); err != errFilterNotFound {
t.Errorf(`expected filter not found error, got "%v"`, err)
}
fid3, err := filterManager.NewBlockFilter()
if err != nil {
t.Fatal(err)
}
for _, h := range blockHashes[cutoff2:] {
ev := core.ChainEvent{Hash: h}
mux.Post(ev)
}
for len(receivedHashes1) < len(blockHashes) {
hashes, err := filterManager.GetBlockFilterChanges(fid1)
if err != nil {
t.Fatalf("unable to fetch block filter changes: %v", err)
}
receivedHashes1 = append(receivedHashes1, hashes...)
}
receivedHashes3, err = filterManager.GetBlockFilterChanges(fid3)
if err != nil {
t.Fatalf("unable to fetch block filter changes: %v", err)
}
if len(receivedHashes1) != len(blockHashes) {
t.Fatalf("received invalid number of block hashes, want %d, got %d", len(blockHashes), len(receivedHashes1))
}
if len(receivedHashes2) != len(blockHashes[cutoff1:cutoff2]) {
t.Fatalf("received invalid number of block hashes, want %d, got %d", len(blockHashes[cutoff1:cutoff2]), len(receivedHashes2))
}
if len(receivedHashes3) != len(blockHashes[cutoff2:]) {
t.Fatalf("received invalid number of block hashes, want %d, got %d", len(blockHashes[cutoff2:]), len(receivedHashes2))
}
// verify hashes
for i, h := range receivedHashes1 {
if blockHashes[i] != h {
t.Errorf("blockhash %d invalid, want %x, got %x", i, blockHashes[i], h)
}
}
for i, h := range receivedHashes2 {
if blockHashes[cutoff1+i] != h {
t.Errorf("blockhash %d invalid, want %x, got %x", i, blockHashes[cutoff1+i], h)
}
}
for i, h := range receivedHashes3 {
if blockHashes[cutoff2+i] != h {
t.Errorf("blockhash %d invalid, want %x, got %x", i, blockHashes[cutoff2+i], h)
}
}
if err := filterManager.Uninstall(fid1); err != nil {
t.Errorf("unable to uninstall %x", fid1)
}
if err := filterManager.Uninstall(fid1); err != errFilterNotFound {
t.Errorf("expected double uninstall to fail, want %s, got %s", errFilterNotFound, err)
}
}
// TestPendingTxFilter tests whether pending tx filters retrieve all pending transactions that are posted to the event mux.
func TestPendingTxFilter(t *testing.T) {
t.Parallel()
var (
transactions = []*types.Transaction{
types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil),
types.NewTransaction(1, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil),
types.NewTransaction(2, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil),
types.NewTransaction(3, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil),
types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil),
}
hashes []common.Hash
)
fid1, err := filterManager.NewPendingTransactionFilter()
if err != nil {
t.Fatal(err)
}
for _, tx := range transactions {
ev := core.TxPreEvent{Tx: tx}
mux.Post(ev)
}
for {
h, err := filterManager.GetPendingTxFilterChanges(fid1)
if err != nil {
t.Fatalf("unable to fetch pending transactions: %v", err)
}
hashes = append(hashes, h...)
if len(hashes) >= len(transactions) {
break
}
time.Sleep(100 * time.Millisecond)
}
for i := range hashes {
if hashes[i] != transactions[i].Hash() {
t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), hashes[i])
}
}
}
// TestLogFilter tests whether log filters match the correct logs that are posted to the event mux.
func TestLogFilter(t *testing.T) {
t.Parallel()
var (
err error
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222")
thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333")
notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999")
firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222")
notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999")
allLogs = vm.Logs{
// Note, these are used for comparison of the test cases.
0: vm.NewLog(firstAddr, []common.Hash{}, []byte(""), 0),
1: vm.NewLog(firstAddr, []common.Hash{firstTopic}, []byte(""), 1),
2: vm.NewLog(secondAddr, []common.Hash{firstTopic}, []byte(""), 1),
3: vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 2),
4: vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 3),
}
testCases = []struct {
crit FilterCriteria
expected vm.Logs
id FilterID
}{
// match all
0: {FilterCriteria{}, allLogs, FilterID{}},
// match none due to no matching addresses
1: {FilterCriteria{Addresses: []common.Address{common.Address{}, notUsedAddress}, Topics: [][]common.Hash{allLogs[0].Topics}}, vm.Logs{}, FilterID{}},
// match logs based on addresses, ignore topics
2: {FilterCriteria{Addresses: []common.Address{firstAddr}}, allLogs[:2], FilterID{}},
// match none due to no matching topics (match with address)
3: {FilterCriteria{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{[]common.Hash{notUsedTopic}}}, vm.Logs{}, FilterID{}},
// match logs based on addresses and topics
4: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, allLogs[3:5], FilterID{}},
// match logs based on multiple addresses and "or" topics
5: {FilterCriteria{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, allLogs[2:5], FilterID{}},
// block numbers are ignored for filters created with New***Filter, these return all logs that match the given criterias when the state changes
6: {FilterCriteria{Addresses: []common.Address{firstAddr}, FromBlock: rpc.BlockNumber(1), ToBlock: rpc.BlockNumber(2)}, allLogs[:2], FilterID{}},
}
)
// create all filters
for i := range testCases {
testCases[i].id, err = filterManager.NewLogFilter(testCases[i].crit, nil)
if err != nil {
t.Fatal(err)
}
}
// raise events
if err = mux.Post(allLogs); err != nil {
t.Fatal(err)
}
for i, tt := range testCases {
var fetched []Log
for { // fetch all expected logs
logs, err := filterManager.GetLogFilterChanges(tt.id)
if err != nil {
t.Errorf("unable to retrieve logs for case %d, %s", i, err)
break
}
fetched = append(fetched, logs...)
if len(fetched) >= len(tt.expected) {
break
}
time.Sleep(100 * time.Millisecond)
}
if len(fetched) != len(tt.expected) {
t.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched))
return
}
for l := range fetched {
if fetched[l].Removed {
t.Errorf("expected log not to be removed for log %d in case %d", l, i)
}
if !reflect.DeepEqual(fetched[l].Log, tt.expected[l]) {
t.Errorf("invalid log on index %d for case %d", l, i)
}
}
}
}
func TestPendingLogFilter(t *testing.T) {
t.Parallel()
var (
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222")
thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333")
notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999")
firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222")
thirdTopic = common.HexToHash("0x3333333333333333333333333333333333333333333333333333333333333333")
forthTopic = common.HexToHash("0x4444444444444444444444444444444444444444444444444444444444444444")
notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999")
allLogs = []core.PendingLogsEvent{
0: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(firstAddr, []common.Hash{}, []byte(""), 0)}},
1: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(firstAddr, []common.Hash{firstTopic}, []byte(""), 1)}},
2: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(secondAddr, []common.Hash{firstTopic}, []byte(""), 2)}},
3: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 3)}},
4: core.PendingLogsEvent{Logs: vm.Logs{vm.NewLog(thirdAddress, []common.Hash{secondTopic}, []byte(""), 4)}},
5: core.PendingLogsEvent{Logs: vm.Logs{
vm.NewLog(thirdAddress, []common.Hash{firstTopic}, []byte(""), 5),
vm.NewLog(thirdAddress, []common.Hash{thirdTopic}, []byte(""), 5),
vm.NewLog(thirdAddress, []common.Hash{forthTopic}, []byte(""), 5),
vm.NewLog(firstAddr, []common.Hash{firstTopic}, []byte(""), 5),
}},
}
concatLogs = func(pl []core.PendingLogsEvent) vm.Logs {
var logs vm.Logs
for _, l := range pl {
logs = append(logs, l.Logs...)
}
return logs
}
testCases = []struct {
crit FilterCriteria
expected vm.Logs
id FilterID
}{
// match all
0: {FilterCriteria{}, concatLogs(allLogs), FilterID{}},
// match none due to no matching addresses
1: {FilterCriteria{Addresses: []common.Address{common.Address{}, notUsedAddress}, Topics: [][]common.Hash{[]common.Hash{}}}, vm.Logs{}, FilterID{}},
// match logs based on addresses, ignore topics
2: {FilterCriteria{Addresses: []common.Address{firstAddr}}, append(concatLogs(allLogs[:2]), allLogs[5].Logs[3]), FilterID{}},
// match none due to no matching topics (match with address)
3: {FilterCriteria{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{[]common.Hash{notUsedTopic}}}, vm.Logs{}, FilterID{}},
// match logs based on addresses and topics
4: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, append(concatLogs(allLogs[3:5]), allLogs[5].Logs[0]), FilterID{}},
// match logs based on multiple addresses and "or" topics
5: {FilterCriteria{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, secondTopic}}}, append(concatLogs(allLogs[2:5]), allLogs[5].Logs[0]), FilterID{}},
// block numbers are ignored for filters created with New***Filter, these return all logs that match the given criterias when the state changes
6: {FilterCriteria{Addresses: []common.Address{firstAddr}, FromBlock: rpc.BlockNumber(2), ToBlock: rpc.BlockNumber(3)}, append(concatLogs(allLogs[:2]), allLogs[5].Logs[3]), FilterID{}},
// multiple pending logs, should match only 2 topics from the logs in block 5
7: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{[]common.Hash{firstTopic, forthTopic}}}, vm.Logs{allLogs[5].Logs[0], allLogs[5].Logs[2]}, FilterID{}},
}
err error
)
// create all filters
for i := range testCases {
testCases[i].id, err = filterManager.NewPendingLogFilter(testCases[i].crit, nil)
if err != nil {
t.Fatal(err)
}
}
// raise events
for _, l := range allLogs {
if err := mux.Post(l); err != nil {
t.Fatal(err)
}
}
for i, tt := range testCases {
var fetched []Log
for {
logs, err := filterManager.GetPendingLogFilterChanges(tt.id)
if err != nil {
t.Errorf("unable to retrieve logs for case %d, %s", i, err)
break
}
fetched = append(fetched, logs...)
if len(fetched) >= len(tt.expected) {
break
}
time.Sleep(100 * time.Millisecond)
}
if len(fetched) != len(tt.expected) {
t.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched))
continue
}
for l := range fetched {
if fetched[l].Removed {
t.Errorf("expected log not to be removed for log %d in case %d", l, i)
}
if !reflect.DeepEqual(fetched[l].Log, tt.expected[l]) {
t.Errorf("invalid log on index %d for case %d", l, i)
}
}
} }
} }