diff --git a/eth/filters/api.go b/eth/filters/api.go
index 393019f8b0..dc4ec4fed5 100644
--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@@ -17,283 +17,191 @@
package filters
import (
- "crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"sync"
- "time"
+
+ "golang.org/x/net/context"
"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/event"
"github.com/ethereum/go-ethereum/rpc"
-
- "golang.org/x/net/context"
)
-var (
- filterTickerTime = 5 * time.Minute
-)
-
-// 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.
+// PublicFilterAPI allows clients to retrieve or get notified of data such as blocks,
+// transactions and logs.
type PublicFilterAPI struct {
- mux *event.TypeMux
+ db ethdb.Database
+ m *Manager
- quit chan struct{}
- chainDb ethdb.Database
-
- 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
+ subscriptionsMu sync.Mutex
+ subscriptions map[string]FilterID // mapping between subscriptions and the underlying filters
}
// 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 {
- svc := &PublicFilterAPI{
- mux: mux,
- chainDb: chainDb,
- filterManager: NewFilterSystem(mux),
- filterMapping: make(map[string]int),
- logQueue: make(map[int]*logQueue),
- blockQueue: make(map[int]*hashQueue),
- transactionQueue: make(map[int]*hashQueue),
+ return &PublicFilterAPI{
+ db: chainDb,
+ m: NewManager(mux),
+ subscriptions: make(map[string]FilterID),
}
- go svc.start()
- return svc
}
-// Stop quits the work loop.
-func (s *PublicFilterAPI) Stop() {
- close(s.quit)
+// NewBlockFilter creates a filter that returns hashes for new blocks.
+// To check if the state has changed, call GetFilterChanges.
+//
+// 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.
-func (s *PublicFilterAPI) start() {
- timer := time.NewTicker(2 * time.Second)
- defer timer.Stop()
-done:
- for {
- 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()
+// NewPendingTransactionFilter creates a filter, to notify when new pending transactions arrive.
+// To check if the state has changed, call GetFilterChanges.
+//
+// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter
+func (api *PublicFilterAPI) NewPendingTransactionFilter() (FilterID, error) {
+ return api.m.NewPendingTransactionFilter()
+}
- s.blockMu.Lock()
- for id, filter := range s.blockQueue {
- if time.Since(filter.timeout) > filterTickerTime {
- s.filterManager.Remove(id)
- delete(s.blockQueue, id)
- }
- }
- s.blockMu.Unlock()
-
- s.transactionMu.Lock()
- for id, filter := range s.transactionQueue {
- if time.Since(filter.timeout) > filterTickerTime {
- s.filterManager.Remove(id)
- delete(s.transactionQueue, id)
- }
- }
- s.transactionMu.Unlock()
- case <-s.quit:
- break done
- }
+// GetFilterChanges returns new logs for the given filter since last time is was called.
+// This can be used for polling.
+//
+// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterchanges
+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.
+ switch api.m.FilterType(id) {
+ case BlockFilter:
+ hashes, err := api.m.GetBlockFilterChanges(id)
+ return toRPCHashes(hashes), err
+ case PendingTxFilter:
+ hashes, err := api.m.GetPendingTxFilterChanges(id)
+ return toRPCHashes(hashes), err
+ case LogFilter:
+ logs, err := api.m.GetLogFilterChanges(id)
+ return toRPCLogs(logs), err
}
+ return nil, errFilterNotFound
}
-// NewBlockFilter create a new filter that returns blocks that are included into the canonical chain.
-func (s *PublicFilterAPI) NewBlockFilter() (string, error) {
- externalId, err := newFilterId()
+// NewFilter creates a filter that can be used to fetch new logs.
+// Note: as specification dictates, it can be used to fetch logs when the state changes.
+// 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 {
- return "", err
+ return nil, err
}
-
- 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
+ return api.GetLogs(crit), nil
}
-// NewPendingTransactionFilter creates a filter that returns new pending transactions.
-func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) {
- externalId, err := newFilterId()
- if err != nil {
- 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) {
+// Logs creates a subscription that fires each time a log is created that matches the given criteria.
+// Note, in case of chain reorganisations logs with the Removed true property can be returned indicating
+// a previously sent log is reverted.
+func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (subscription rpc.Subscription, err error) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return nil, rpc.ErrNotificationsUnsupported
}
- var (
- externalId string
- subscription rpc.Subscription
- err error
- )
-
- 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()
+ onUnsubscribe := func(subid string) {
+ api.subscriptionsMu.Lock()
+ defer api.subscriptionsMu.Unlock()
+ if fid, found := api.subscriptions[subid]; found {
+ delete(api.subscriptions, subid)
+ api.m.Uninstall(fid) // uninstall associated filter
}
}
- // from and to block number are not used since subscriptions don't allow you to travel to "time"
- 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)
- }
-
+ subscription, err = notifier.NewSubscription(onUnsubscribe)
if err != nil {
- subscription.Cancel()
return nil, err
}
- s.filterMapMu.Lock()
- s.filterMapping[externalId] = id
- s.filterMapMu.Unlock()
+ send := func(id FilterID, logs []Log) {
+ rpcLogs := toRPCLogs(logs)
+ 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
}
-// NewFilterArgs represents a request to create a new filter.
-type NewFilterArgs struct {
+// UninstallFilter deletes a filter by its id. If successful true is returned,
+// 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
ToBlock rpc.BlockNumber
Addresses []common.Address
Topics [][]common.Hash
}
-// UnmarshalJSON sets *args fields with given data.
-func (args *NewFilterArgs) UnmarshalJSON(data []byte) error {
+// UnmarshalJSON sets *args fields with filter criteria JSON encoded in the given data.
+func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
type input struct {
From *rpc.BlockNumber `json:"fromBlock"`
ToBlock *rpc.BlockNumber `json:"toBlock"`
@@ -405,245 +313,3 @@ func (args *NewFilterArgs) UnmarshalJSON(data []byte) error {
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
-}
diff --git a/eth/filters/api_test.go b/eth/filters/api_test.go
index 9e8edc2413..81c7105bb6 100644
--- a/eth/filters/api_test.go
+++ b/eth/filters/api_test.go
@@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see .
-package filters_test
+package filters
import (
"encoding/json"
@@ -22,7 +22,6 @@ import (
"testing"
"github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/eth/filters"
"github.com/ethereum/go-ethereum/rpc"
)
@@ -39,7 +38,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
)
// default values
- var test0 filters.NewFilterArgs
+ var test0 FilterCriteria
if err := json.Unmarshal([]byte("{}"), &test0); err != nil {
t.Fatal(err)
}
@@ -57,7 +56,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// from, to block number
- var test1 filters.NewFilterArgs
+ var test1 FilterCriteria
vector := fmt.Sprintf(`{"fromBlock":"0x%x","toBlock":"0x%x"}`, fromBlock, toBlock)
if err := json.Unmarshal([]byte(vector), &test1); err != nil {
t.Fatal(err)
@@ -70,7 +69,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// single address
- var test2 filters.NewFilterArgs
+ var test2 FilterCriteria
vector = fmt.Sprintf(`{"address": "%s"}`, address0.Hex())
if err := json.Unmarshal([]byte(vector), &test2); err != nil {
t.Fatal(err)
@@ -83,7 +82,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// multiple address
- var test3 filters.NewFilterArgs
+ var test3 FilterCriteria
vector = fmt.Sprintf(`{"address": ["%s", "%s"]}`, address0.Hex(), address1.Hex())
if err := json.Unmarshal([]byte(vector), &test3); err != nil {
t.Fatal(err)
@@ -99,7 +98,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// single topic
- var test4 filters.NewFilterArgs
+ var test4 FilterCriteria
vector = fmt.Sprintf(`{"topics": ["%s"]}`, topic0.Hex())
if err := json.Unmarshal([]byte(vector), &test4); err != nil {
t.Fatal(err)
@@ -115,7 +114,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// test multiple "AND" topics
- var test5 filters.NewFilterArgs
+ var test5 FilterCriteria
vector = fmt.Sprintf(`{"topics": ["%s", "%s"]}`, topic0.Hex(), topic1.Hex())
if err := json.Unmarshal([]byte(vector), &test5); err != nil {
t.Fatal(err)
@@ -137,7 +136,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// test optional topic
- var test6 filters.NewFilterArgs
+ var test6 FilterCriteria
vector = fmt.Sprintf(`{"topics": ["%s", null, "%s"]}`, topic0.Hex(), topic2.Hex())
if err := json.Unmarshal([]byte(vector), &test6); err != nil {
t.Fatal(err)
@@ -165,7 +164,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
}
// 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())
if err := json.Unmarshal([]byte(vector), &test7); err != nil {
t.Fatal(err)
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index 995b588fb2..821bc6736e 100644
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@@ -23,15 +23,10 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/ethdb"
)
-type AccountChange struct {
- Address, StateAddress []byte
-}
-
-// Filtering interface
+// Filter can be used to retrieve and filter logs
type Filter struct {
created time.Time
@@ -39,67 +34,67 @@ type Filter struct {
begin, end int64
addresses []common.Address
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
-// is interesting or not.
+// New creates a new filter which uses a bloom filter on blocks to figure out whether
+// a particular block is interesting or not.
func New(db ethdb.Database) *Filter {
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)
// hash = particular hash from-to
-func (self *Filter) SetBeginBlock(begin int64) {
- self.begin = begin
+func (f *Filter) SetBeginBlock(begin int64) {
+ f.begin = begin
}
-func (self *Filter) SetEndBlock(end int64) {
- self.end = end
+// SetEndBlock sets the latest block for filtering.
+func (f *Filter) SetEndBlock(end int64) {
+ f.end = end
}
-func (self *Filter) SetAddresses(addr []common.Address) {
- self.addresses = addr
+// SetAddresses matches only logs that are generated from addresses that are included
+// in the given addresses.
+func (f *Filter) SetAddresses(addr []common.Address) {
+ f.addresses = addr
}
-func (self *Filter) SetTopics(topics [][]common.Hash) {
- self.topics = topics
+// SetTopics matches only logs that have topics matching the given topics.
+func (f *Filter) SetTopics(topics [][]common.Hash) {
+ f.topics = topics
}
-// Run filters logs with the current parameters set
-func (self *Filter) Find() vm.Logs {
- latestHash := core.GetHeadBlockHash(self.db)
- latestBlock := core.GetBlock(self.db, latestHash, core.GetBlockNumber(self.db, latestHash))
- var beginBlockNo uint64 = uint64(self.begin)
- if self.begin == -1 {
+// Find logs with the current parameters set.
+func (f *Filter) Find() []Log {
+ latestHash := core.GetHeadBlockHash(f.db)
+ latestBlock := core.GetBlock(f.db, latestHash, core.GetBlockNumber(f.db, latestHash))
+ beginBlockNo := uint64(f.begin)
+ if f.begin == -1 {
beginBlockNo = latestBlock.NumberU64()
}
- var endBlockNo uint64 = uint64(self.end)
- if self.end == -1 {
+ endBlockNo := uint64(f.end)
+ if f.end == -1 {
endBlockNo = latestBlock.NumberU64()
}
// 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
// higher range probability in order to ensure at least a false positive
- if len(self.addresses) == 0 {
- return self.getLogs(beginBlockNo, endBlockNo)
+ if len(f.addresses) == 0 {
+ 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]
// normalise numerator so we can work in level specific batches and
// work with the proper range checks
for num := start / level * level; num <= end; num += level {
// find addresses in bloom filters
- bloom := core.GetMipmapBloom(self.db, num, level)
- for _, addr := range self.addresses {
+ bloom := core.GetMipmapBloom(f.db, num, level)
+ for _, addr := range f.addresses {
if bloom.TestBytes(addr[:]) {
// range check normalised values and make sure that
// 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)))
end := uint64(math.Min(float64(num+level-1), float64(end)))
if depth+1 == len(core.MIPMapLevels) {
- logs = append(logs, self.getLogs(start, end)...)
+ logs = append(logs, f.getLogs(start, end)...)
} 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
// possible address. Checks on multiple addresses
@@ -122,29 +117,33 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.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
for i := start; i <= end; i++ {
- hash := core.GetCanonicalHash(self.db, i)
+ hash := core.GetCanonicalHash(f.db, i)
if hash != (common.Hash{}) {
- block = core.GetBlock(self.db, hash, i)
+ block = core.GetBlock(f.db, hash, i)
} else { // block not found
return logs
}
// Use bloom filtering to see if this block is interesting given the
// current parameters
- if self.bloomFilter(block) {
+ if f.bloomFilter(block) {
// Get the logs of the block
var (
- receipts = core.GetBlockReceipts(self.db, block.Hash(), i)
- unfiltered vm.Logs
+ receipts = core.GetBlockReceipts(f.db, block.Hash(), i)
+ unfiltered []Log
)
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
}
-func (self *Filter) FilterLogs(logs vm.Logs) vm.Logs {
- var ret vm.Logs
+func filterLogs(logs []Log, addresses []common.Address, topics [][]common.Hash) []Log {
+ var ret []Log
// Filter the logs for interesting stuff
Logs:
for _, log := range logs {
- if len(self.addresses) > 0 && !includes(self.addresses, log.Address) {
+ if len(addresses) > 0 && !includes(addresses, log.Address) {
continue
}
- logTopics := make([]common.Hash, len(self.topics))
+ logTopics := make([]common.Hash, len(topics))
copy(logTopics, log.Topics)
// If the to filtered topics is greater than the amount of topics in
// logs, skip.
- if len(self.topics) > len(log.Topics) {
+ if len(topics) > len(log.Topics) {
continue Logs
}
- for i, topics := range self.topics {
+ for i, topics := range topics {
var match bool
for _, topic := range topics {
// common.Hash{} is a match all (wildcard)
@@ -202,10 +201,10 @@ Logs:
return ret
}
-func (self *Filter) bloomFilter(block *types.Block) bool {
- if len(self.addresses) > 0 {
+func (f *Filter) bloomFilter(block *types.Block) bool {
+ if len(f.addresses) > 0 {
var included bool
- for _, addr := range self.addresses {
+ for _, addr := range f.addresses {
if types.BloomLookup(block.Bloom(), addr) {
included = true
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
for _, topic := range sub {
if (topic == common.Hash{}) || types.BloomLookup(block.Bloom(), topic) {
diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go
index 4343dfa21c..c2cf47a6b7 100644
--- a/eth/filters/filter_system.go
+++ b/eth/filters/filter_system.go
@@ -14,172 +14,576 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see .
-// package filters implements an ethereum filtering system for block,
+// Package filters implements an ethereum filtering system for block,
// transactions and log events.
package filters
import (
+ "bufio"
+ crand "crypto/rand"
+ "encoding/binary"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
"fmt"
+ "math/rand"
"sync"
"time"
+ "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/vm"
"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 (
- ChainFilter FilterType = iota // new block events filter
- PendingTxFilter // pending transaction filter
- LogFilter // new or removed log filter
- PendingLogFilter // pending log filter
+ // UnknownFilter indicates an unkown filter type
+ UnknownFilter Type = iota
+ // BlockFilter queries for new blocks
+ 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
-// block, transaction and log events. The Filtering system can be used to listen
-// for specific LOG events fired by the EVM (Ethereum Virtual Machine).
-type FilterSystem struct {
- filterMu sync.RWMutex
- filterId int
+var (
+ errFilterNotFound = errors.New("filter not found")
+ errCouldNotGenerateFilterID = errors.New("unable to generate filter id")
+ errInvalidFilterID = errors.New("invalid filter id")
+ errUnableToUninstallFilter = errors.New("unable to uninstall filter, retry later")
- chainFilters map[int]*Filter
- pendingTxFilters map[int]*Filter
- logFilters map[int]*Filter
- pendingLogFilters map[int]*Filter
+ filterIDGenMu sync.Mutex
+ filterIDGen = filterIDGenerator()
+)
- // generic is an ugly hack for Get
- generic map[int]*Filter
+type logsCallback func(FilterID, []Log)
- 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
-func NewFilterSystem(mux *event.TypeMux) *FilterSystem {
- fs := &FilterSystem{
- chainFilters: make(map[int]*Filter),
- pendingTxFilters: make(map[int]*Filter),
- logFilters: make(map[int]*Filter),
- pendingLogFilters: make(map[int]*Filter),
- generic: make(map[int]*Filter),
+// filterIDGenerator helper utility that generates a (pseudo) random sequence of bytes
+// that are used to generate filter identifiers.
+func filterIDGenerator() *rand.Rand {
+ if seed, err := binary.ReadVarint(bufio.NewReader(crand.Reader)); err == nil {
+ return rand.New(rand.NewSource(seed))
}
- 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.RemovedLogsEvent{},
core.ChainEvent{},
core.TxPreEvent{},
vm.Logs(nil),
)
- go fs.filterLoop()
- return fs
-}
-// Stop quits the filter loop required for polling events
-func (fs *FilterSystem) Stop() {
- fs.sub.Unsubscribe()
-}
-
-// 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)
+ m := &Manager{
+ sub: sub,
+ install: make(chan *filter),
+ uninstall: make(chan *filter, 1024),
+ all: make(map[FilterID]*filter),
}
- 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
}
-// Remove removes a filter by filter id
-func (fs *FilterSystem) Remove(id int) {
- fs.filterMu.Lock()
- defer fs.filterMu.Unlock()
+// GetBlockFilterChanges returns the data for the filter with the given id since
+// last time it was called.
+func (m *Manager) GetBlockFilterChanges(id FilterID) ([]common.Hash, error) {
+ m.allMu.RLock()
+ if f, found := m.all[id]; found && f.typ == BlockFilter {
+ m.allMu.RUnlock()
- delete(fs.chainFilters, id)
- delete(fs.pendingTxFilters, id)
- delete(fs.logFilters, id)
- delete(fs.pendingLogFilters, id)
- delete(fs.generic, id)
+ // retieve hashes
+ f.lastUsed = time.Now()
+ hashes := make([]common.Hash, 0, len(f.hashes)) // prevent (most) allocs
+ for {
+ 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 {
- fs.filterMu.RLock()
- defer fs.filterMu.RUnlock()
+// NewLogFilterWithNoTimeout returns a filter identifier that can be used to fetch
+// logs matching the given criteria. The created filter will not timeout and the
+// 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
-// when the filter matches the requirements.
-func (fs *FilterSystem) filterLoop() {
- for event := range fs.sub.Chan() {
- 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()
+// NewLogFilter returns a filter identifier that can be used to fetch logs matching
+// the given criteria.
+func (m *Manager) NewLogFilter(crit FilterCriteria, cb logsCallback) (FilterID, error) {
+ id := newID()
- case vm.Logs:
- fs.filterMu.RLock()
- for _, filter := range fs.logFilters {
- if filter.LogCallback != nil && !filter.created.After(event.Time) {
- for _, log := range filter.FilterLogs(ev) {
- filter.LogCallback(log, false)
- }
+ f := &filter{
+ ID: id,
+ created: time.Now(),
+ lastUsed: time.Now(),
+ canTimeout: true,
+ 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:
- fs.filterMu.RLock()
- for _, filter := range fs.logFilters {
- if filter.LogCallback != nil && !filter.created.After(event.Time) {
- for _, removedLog := range filter.FilterLogs(ev.Logs) {
- filter.LogCallback(removedLog, true)
- }
+ }
+ case core.TxPreEvent:
+ for _, f := range filters[PendingTxFilter] {
+ if ev.Time.After(f.created) {
+ select {
+ case f.hashes <- e.Tx.Hash():
+ continue
+ default:
+ // data queue full, disable filter
+ inactive = append(inactive, f)
}
}
- fs.filterMu.RUnlock()
- case core.PendingLogsEvent:
- fs.filterMu.RLock()
- for _, filter := range fs.pendingLogFilters {
- if filter.LogCallback != nil && !filter.created.After(event.Time) {
- for _, pendingLog := range ev.Logs {
- filter.LogCallback(pendingLog, false)
- }
- }
+ }
+ case vm.Logs:
+ for _, f := range filters[LogFilter] {
+ if ev.Time.After(f.created) {
+ matchedLogs := filterLogs(convertLogs(e, false), f.logsCrit.Addresses, f.logsCrit.Topics)
+ logHandler(f, matchedLogs)
}
- 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
+}
diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go
index 72824cb088..5cbf51a132 100644
--- a/eth/filters/filter_system_test.go
+++ b/eth/filters/filter_system_test.go
@@ -17,101 +17,442 @@
package filters
import (
+ "encoding/json"
+ "math/big"
+ "reflect"
"testing"
"time"
+ "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/core/vm"
"github.com/ethereum/go-ethereum/event"
+ "github.com/ethereum/go-ethereum/rpc"
)
-func TestCallbacks(t *testing.T) {
- var (
- mux event.TypeMux
- fs = NewFilterSystem(&mux)
- blockDone = make(chan struct{})
- txDone = make(chan struct{})
- logDone = make(chan struct{})
- removedLogDone = make(chan struct{})
- pendingLogDone = make(chan struct{})
- )
+var (
+ mux = new(event.TypeMux)
+ filterManager = NewManager(mux)
+)
- blockFilter := &Filter{
- BlockCallback: func(*types.Block, vm.Logs) {
- close(blockDone)
- },
- }
- txFilter := &Filter{
- TransactionCallback: func(*types.Transaction) {
- close(txDone)
- },
- }
- 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)
- },
+// TestFilterIdSerialization tests if FilterIs is correct serialized and parsed.
+func TestFilterIdSerialization(t *testing.T) {
+ t.Parallel()
+
+ id := newID()
+
+ serialized, err := json.Marshal(id)
+ if err != nil {
+ t.Fatal(err)
}
- fs.Add(blockFilter, ChainFilter)
- fs.Add(txFilter, PendingTxFilter)
- 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)")
+ if len(serialized) != 36 {
+ t.Fatalf("invalid filter id length (%s), want %d, got %d", serialized, 36, len(serialized))
}
- failTimer.Reset(dura)
- select {
- case <-txDone:
- case <-failTimer.C:
- t.Error("transaction filter failed to trigger (timeout)")
+ var filterID FilterID
+ if err := json.Unmarshal(serialized, &filterID); err != nil {
+ t.Fatal(err)
}
- failTimer.Reset(dura)
- select {
- case <-logDone:
- case <-failTimer.C:
- t.Error("log filter failed to trigger (timeout)")
- }
-
- failTimer.Reset(dura)
- select {
- case <-removedLogDone:
- case <-failTimer.C:
- t.Error("removed log filter failed to trigger (timeout)")
- }
-
- failTimer.Reset(dura)
- select {
- case <-pendingLogDone:
- case <-failTimer.C:
- t.Error("pending log filter failed to trigger (timeout)")
+ if filterID != id {
+ t.Errorf("invalid filter id, want %x, got %x", id, filterID)
+ }
+}
+
+// TestBlockFilter tests if a block filter callback is called when core.ChainEvent are posted.
+// It creates multiple filters:
+// - one at the start and should receive all posted chain events and a second (blockHashes)
+// - one that is created after a cutoff moment and uninstalled after a second cutoff moment (blockHashes[cutoff1:cutoff2])
+// - one that is created after the second cutoff moment (blockHashes[cutoff2:])
+func TestBlockFilter(t *testing.T) {
+ t.Parallel()
+
+ var (
+ blockHashes = []common.Hash{
+ common.HexToHash("0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"),
+ common.HexToHash("0x88e96d4537bea4d9c05d12549907b32561d3bf31f45aae734cdc119f13406cb6"),
+ common.HexToHash("0xb495a1d7e6663152ae92708da4843337b958146015a2802f4193a410044698c9"),
+ 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)
+ }
+ }
}
}