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

Merge f99cc5a840 into a4c4125b11

Browse source
This commit is contained in:
bas-vk 2016-07-16 12:30:00 +00:00 committed by GitHub
commit c1c1c7cd38
5 changed files with 1138 additions and 729 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

608
eth/filters/api.go
View file

@ -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),
}
go svc.start()
return svc
}
// Stop quits the work loop.
func (s *PublicFilterAPI) Stop() {
close(s.quit)
}
// 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()
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
return &PublicFilterAPI{
db: chainDb,
m: NewManager(mux),
subscriptions: make(map[string]FilterID),
}
}
// 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()
}
// NewBlockFilter create a new filter that returns blocks that are included into the canonical chain.
func (s *PublicFilterAPI) NewBlockFilter() (string, error) {
externalId, err := newFilterId()
// 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()
}
// 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
}
// 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
}
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.
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
}

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
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
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)

109
eth/filters/filter.go
View file

@ -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}
}
logs = append(logs, self.FilterLogs(unfiltered)...)
unfiltered = append(unfiltered, rl...)
}
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) {

612
eth/filters/filter_system.go
View file

@ -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 <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.
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)
// 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))
}
// 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))
}
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
}
// 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),
}
fs.sub = mux.Subscribe(
// 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
m := &Manager{
sub: sub,
install: make(chan *filter),
uninstall: make(chan *filter, 1024),
all: make(map[FilterID]*filter),
}
// Stop quits the filter loop required for polling events
func (fs *FilterSystem) Stop() {
fs.sub.Unsubscribe()
go m.run()
return m
}
// 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()
// Uninstall filter. If the given filter could not be found an error is returned.
func (m *Manager) Uninstall(id FilterID) error {
m.allMu.Lock()
id := fs.filterId
filter.created = time.Now()
if f, found := m.all[id]; found {
delete(m.all, id)
m.allMu.Unlock()
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
select {
case m.uninstall <- f:
return nil
default:
return 0, fmt.Errorf("unknown filter type %v", filterType)
// can only happen when there are too many pending uninstall requests
return errUnableToUninstallFilter
}
}
fs.generic[id] = filter
fs.filterId++
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
}
}
}
func (fs *FilterSystem) Get(id int) *Filter {
fs.filterMu.RLock()
defer fs.filterMu.RUnlock()
return fs.generic[id]
m.allMu.RUnlock()
return nil, errFilterNotFound
}
// 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) {
// 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()
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
}
// 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()
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:
fs.filterMu.RLock()
for _, filter := range fs.chainFilters {
if filter.BlockCallback != nil && !filter.created.After(event.Time) {
filter.BlockCallback(ev.Block, ev.Logs)
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.TxPreEvent:
fs.filterMu.RLock()
for _, filter := range fs.pendingTxFilters {
if filter.TransactionCallback != nil && !filter.created.After(event.Time) {
filter.TransactionCallback(ev.Tx)
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 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)
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:
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)
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)
}
}
}
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)
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)
}
}
}
fs.filterMu.RUnlock()
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
}

489
eth/filters/filter_system_test.go
View file

@ -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{})
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)")
if filterID != id {
t.Errorf("invalid filter id, want %x, got %x", id, filterID)
}
}
failTimer.Reset(dura)
select {
case <-removedLogDone:
case <-failTimer.C:
t.Error("removed log filter failed to trigger (timeout)")
// 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"),
}
failTimer.Reset(dura)
select {
case <-pendingLogDone:
case <-failTimer.C:
t.Error("pending log filter failed to trigger (timeout)")
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)
}
}
}
}