diff --git a/eth/filters/api.go b/eth/filters/api.go index aef9adba103f..9b7c6fb6fbaf 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -18,43 +18,23 @@ package filters import ( "context" - "encoding/json" "errors" "fmt" "math/big" - "sync" "time" - ethereum "github.com/XinFinOrg/XDPoSChain" "github.com/XinFinOrg/XDPoSChain/common" "github.com/XinFinOrg/XDPoSChain/common/hexutil" - "github.com/XinFinOrg/XDPoSChain/core" "github.com/XinFinOrg/XDPoSChain/core/types" "github.com/XinFinOrg/XDPoSChain/ethdb" - "github.com/XinFinOrg/XDPoSChain/event" "github.com/XinFinOrg/XDPoSChain/rpc" ) -var ( - deadline = 5 * time.Minute // consider a filter inactive if it has not been polled for within deadline -) - -// filter is a helper struct that holds meta information over the filter type -// and associated subscription in the event system. -type filter struct { - typ Type - deadline *time.Timer // filter is inactiv when deadline triggers - hashes []common.Hash - crit FilterCriteria - logs []*types.Log - s *Subscription // associated subscription in event system -} - // 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. +// information related to the Ethereum protocol such as blocks, transactions and logs. type PublicFilterAPI struct { backend Backend - mux *event.TypeMux + mux *EventSystem quit chan struct{} chainDb ethdb.Database events *EventSystem @@ -66,9 +46,9 @@ type PublicFilterAPI struct { func NewPublicFilterAPI(backend Backend, lightMode bool) *PublicFilterAPI { api := &PublicFilterAPI{ backend: backend, - mux: backend.EventMux(), + mux: NewEventSystem(backend, lightMode), chainDb: backend.ChainDb(), - events: NewEventSystem(backend.EventMux(), backend, lightMode), + events: NewEventSystem(backend, lightMode), filters: make(map[rpc.ID]*filter), } go api.timeoutLoop() @@ -76,367 +56,39 @@ func NewPublicFilterAPI(backend Backend, lightMode bool) *PublicFilterAPI { return api } -// timeoutLoop runs every 5 minutes and deletes filters that have not been recently used. -// Tt is started when the api is created. -func (api *PublicFilterAPI) timeoutLoop() { - ticker := time.NewTicker(5 * time.Minute) - for { - <-ticker.C - api.filtersMu.Lock() - for id, f := range api.filters { - select { - case <-f.deadline.C: - f.s.Unsubscribe() - delete(api.filters, id) - default: - continue - } - } - api.filtersMu.Unlock() - } -} - -// NewPendingTransactionFilter creates a filter that fetches pending transaction hashes -// as transactions enter the pending state. -// -// It is part of the filter package because this filter can be used throug the -// `eth_getFilterChanges` polling method that is also used for log filters. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter -func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID { - var ( - pendingTxs = make(chan common.Hash) - pendingTxSub = api.events.SubscribePendingTxEvents(pendingTxs) - ) - - api.filtersMu.Lock() - api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: pendingTxSub} - api.filtersMu.Unlock() - - go func() { - for { - select { - case ph := <-pendingTxs: - api.filtersMu.Lock() - if f, found := api.filters[pendingTxSub.ID]; found { - f.hashes = append(f.hashes, ph) - } - api.filtersMu.Unlock() - case <-pendingTxSub.Err(): - api.filtersMu.Lock() - delete(api.filters, pendingTxSub.ID) - api.filtersMu.Unlock() - return - } - } - }() - - return pendingTxSub.ID -} - -// NewPendingTransactions creates a subscription that is triggered each time a transaction -// enters the transaction pool and was signed from one of the transactions this nodes manages. -func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - txHashes := make(chan common.Hash) - pendingTxSub := api.events.SubscribePendingTxEvents(txHashes) - - for { - select { - case h := <-txHashes: - notifier.Notify(rpcSub.ID, h) - case <-rpcSub.Err(): - pendingTxSub.Unsubscribe() - return - case <-notifier.Closed(): - pendingTxSub.Unsubscribe() - return - } - } - }() - - return rpcSub, nil -} - -// NewBlockFilter creates a filter that fetches blocks that are imported into the chain. -// It is part of the filter package since polling goes with eth_getFilterChanges. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newblockfilter -func (api *PublicFilterAPI) NewBlockFilter() rpc.ID { - var ( - headers = make(chan *types.Header) - headerSub = api.events.SubscribeNewHeads(headers) - ) - - api.filtersMu.Lock() - api.filters[headerSub.ID] = &filter{typ: BlocksSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: headerSub} - api.filtersMu.Unlock() - - go func() { - for { - select { - case h := <-headers: - api.filtersMu.Lock() - if f, found := api.filters[headerSub.ID]; found { - f.hashes = append(f.hashes, h.Hash()) - } - api.filtersMu.Unlock() - case <-headerSub.Err(): - api.filtersMu.Lock() - delete(api.filters, headerSub.ID) - api.filtersMu.Unlock() - return - } - } - }() - - return headerSub.ID -} - -// NewHeads send a notification each time a new (header) block is appended to the chain. -func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - headers := make(chan *types.Header) - headersSub := api.events.SubscribeNewHeads(headers) - - for { - select { - case h := <-headers: - notifier.Notify(rpcSub.ID, h) - case <-rpcSub.Err(): - headersSub.Unsubscribe() - return - case <-notifier.Closed(): - headersSub.Unsubscribe() - return - } - } - }() - - return rpcSub, nil -} - -// Logs creates a subscription that fires for all new log that match the given filter criteria. -func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - var ( - rpcSub = notifier.CreateSubscription() - matchedLogs = make(chan []*types.Log) - ) - - logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), matchedLogs) - if err != nil { - return nil, err - } - - go func() { - - for { - select { - case logs := <-matchedLogs: - for _, log := range logs { - notifier.Notify(rpcSub.ID, &log) - } - case <-rpcSub.Err(): // client send an unsubscribe request - logsSub.Unsubscribe() - return - case <-notifier.Closed(): // connection dropped - logsSub.Unsubscribe() - return - } - } - }() - - return rpcSub, nil -} - -// FilterCriteria represents a request to create a new filter. -// -// TODO(karalabe): Kill this in favor of ethereum.FilterQuery. -type FilterCriteria struct { - FromBlock *big.Int - ToBlock *big.Int - Addresses []common.Address - Topics [][]common.Hash -} - -// NewFilter creates a new filter and returns the filter id. It can be -// used to retrieve logs when the state changes. This method cannot be -// used to fetch logs that are already stored in the state. -// -// Default criteria for the from and to block are "latest". -// Using "latest" as block number will return logs for mined blocks. -// Using "pending" as block number returns logs for not yet mined (pending) blocks. -// In case logs are removed (chain reorg) previously returned logs are returned -// again but with the removed property set to true. -// -// In case "fromBlock" > "toBlock" an error is returned. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newfilter -func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) { - logs := make(chan []*types.Log) - logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), logs) - if err != nil { - return rpc.ID(""), err - } - - api.filtersMu.Lock() - api.filters[logsSub.ID] = &filter{typ: LogsSubscription, crit: crit, deadline: time.NewTimer(deadline), logs: make([]*types.Log, 0), s: logsSub} - api.filtersMu.Unlock() - - go func() { - for { - select { - case l := <-logs: - api.filtersMu.Lock() - if f, found := api.filters[logsSub.ID]; found { - f.logs = append(f.logs, l...) - } - api.filtersMu.Unlock() - case <-logsSub.Err(): - api.filtersMu.Lock() - delete(api.filters, logsSub.ID) - api.filtersMu.Unlock() - return - } - } - }() - - return logsSub.ID, nil -} - // GetLogs returns logs matching the given argument that are stored within the state. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) { - // Convert the RPC block numbers into internal representations - if crit.FromBlock == nil { - crit.FromBlock = big.NewInt(rpc.LatestBlockNumber.Int64()) - } - if crit.ToBlock == nil { - crit.ToBlock = big.NewInt(rpc.LatestBlockNumber.Int64()) - } - // Create and run the filter to get all the logs - filter := New(api.backend, crit.FromBlock.Int64(), crit.ToBlock.Int64(), crit.Addresses, crit.Topics) - - logs, err := filter.Logs(ctx) - if err != nil { - return nil, err - } - return returnLogs(logs), err -} - -// UninstallFilter removes the filter with the given filter id. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_uninstallfilter -func (api *PublicFilterAPI) UninstallFilter(id rpc.ID) bool { - api.filtersMu.Lock() - f, found := api.filters[id] - if found { - delete(api.filters, id) - } - api.filtersMu.Unlock() - if found { - f.s.Unsubscribe() - } - - return found -} - -// GetFilterLogs returns the logs for the filter with the given id. -// If the filter could not be found an empty array of logs is returned. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterlogs -func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*types.Log, error) { - api.filtersMu.Lock() - f, found := api.filters[id] - api.filtersMu.Unlock() - - if !found || f.typ != LogsSubscription { - return nil, fmt.Errorf("filter not found") - } - - begin := rpc.LatestBlockNumber.Int64() - if f.crit.FromBlock != nil { - begin = f.crit.FromBlock.Int64() - } - end := rpc.LatestBlockNumber.Int64() - if f.crit.ToBlock != nil { - end = f.crit.ToBlock.Int64() + var filter *Filter + if crit.BlockHash != nil { + filter = NewBlockFilter(api.backend, *crit.BlockHash, crit.Addresses, crit.Topics) + } else { + begin := rpc.LatestBlockNumber.Int64() + if crit.FromBlock != nil { + begin = crit.FromBlock.Int64() + } + end := rpc.LatestBlockNumber.Int64() + if crit.ToBlock != nil { + end = crit.ToBlock.Int64() + } + filter = NewRangeFilter(api.backend, begin, end, crit.Addresses, crit.Topics) } - // Create and run the filter to get all the logs - filter := New(api.backend, begin, end, f.crit.Addresses, f.crit.Topics) logs, err := filter.Logs(ctx) if err != nil { return nil, err } - return returnLogs(logs), nil -} - -// GetFilterChanges returns the logs for the filter with the given id since -// last time it was called. This can be used for polling. -// -// For pending transaction and block filters the result is []common.Hash. -// (pending)Log filters return []Log. -// -// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterchanges -func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) { - api.filtersMu.Lock() - defer api.filtersMu.Unlock() - - if f, found := api.filters[id]; found { - if !f.deadline.Stop() { - // timer expired but filter is not yet removed in timeout loop - // receive timer value and reset timer - <-f.deadline.C - } - f.deadline.Reset(deadline) - switch f.typ { - case PendingTransactionsSubscription, BlocksSubscription: - hashes := f.hashes - f.hashes = nil - return returnHashes(hashes), nil - case LogsSubscription: - logs := f.logs - f.logs = nil - for _, log := range logs { - // update BlockHash to fix #208 - log.BlockHash = core.GetCanonicalHash(api.chainDb, log.BlockNumber) + // Ensure each log has the correct blockHash from the actual block + for i := range logs { + if logs[i].BlockNumber > 0 { + block := api.backend.GetBlockByNumber(ctx, rpc.BlockNumber(logs[i].BlockNumber)) + if block != nil { + logs[i].BlockHash = block.Hash() } - return returnLogs(logs), nil } } - return []interface{}{}, fmt.Errorf("filter not found") -} - -// returnHashes is a helper that will return an empty hash array case the given hash array is nil, -// otherwise the given hashes array is returned. -func returnHashes(hashes []common.Hash) []common.Hash { - if hashes == nil { - return []common.Hash{} - } - return hashes + return returnLogs(logs), err } // returnLogs is a helper that will return an empty log array in case the given logs array is nil, @@ -447,118 +99,3 @@ func returnLogs(logs []*types.Log) []*types.Log { } return logs } - -// UnmarshalJSON sets *args fields with given data. -func (args *FilterCriteria) UnmarshalJSON(data []byte) error { - type input struct { - From *rpc.BlockNumber `json:"fromBlock"` - ToBlock *rpc.BlockNumber `json:"toBlock"` - Addresses interface{} `json:"address"` - Topics []interface{} `json:"topics"` - } - - var raw input - if err := json.Unmarshal(data, &raw); err != nil { - return err - } - - if raw.From != nil { - args.FromBlock = big.NewInt(raw.From.Int64()) - } - - if raw.ToBlock != nil { - args.ToBlock = big.NewInt(raw.ToBlock.Int64()) - } - - args.Addresses = []common.Address{} - - if raw.Addresses != nil { - // raw.Address can contain a single address or an array of addresses - switch rawAddr := raw.Addresses.(type) { - case []interface{}: - for i, addr := range rawAddr { - if strAddr, ok := addr.(string); ok { - addr, err := decodeAddress(strAddr) - if err != nil { - return fmt.Errorf("invalid address at index %d: %v", i, err) - } - args.Addresses = append(args.Addresses, addr) - } else { - return fmt.Errorf("non-string address at index %d", i) - } - } - case string: - addr, err := decodeAddress(rawAddr) - if err != nil { - return fmt.Errorf("invalid address: %v", err) - } - args.Addresses = []common.Address{addr} - default: - return errors.New("invalid addresses in query") - } - } - - // topics is an array consisting of strings and/or arrays of strings. - // JSON null values are converted to common.Hash{} and ignored by the filter manager. - if len(raw.Topics) > 0 { - args.Topics = make([][]common.Hash, len(raw.Topics)) - for i, t := range raw.Topics { - switch topic := t.(type) { - case nil: - // ignore topic when matching logs - - case string: - // match specific topic - top, err := decodeTopic(topic) - if err != nil { - return err - } - args.Topics[i] = []common.Hash{top} - - case []interface{}: - // or case e.g. [null, "topic0", "topic1"] - for _, rawTopic := range topic { - if rawTopic == nil { - // null component, match all - args.Topics[i] = nil - break - } - if topic, ok := rawTopic.(string); ok { - parsed, err := decodeTopic(topic) - if err != nil { - return err - } - args.Topics[i] = append(args.Topics[i], parsed) - } else { - return fmt.Errorf("invalid topic(s)") - } - } - default: - return fmt.Errorf("invalid topic(s)") - } - } - } - - return nil -} -func hasXDCPrefix(str string) bool { - return len(str) >= 3 && (str[0] == 'x' || str[0] == 'X') && (str[1] == 'd' || str[1] == 'D') && (str[2] == 'c' || str[2] == 'C') -} -func decodeAddress(s string) (common.Address, error) { - if hasXDCPrefix(s) { - s = "0x" + s[3:] - } - b, err := hexutil.Decode(s) - if err == nil && len(b) != common.AddressLength { - err = fmt.Errorf("hex has invalid length %d after decoding", len(b)) - } - return common.BytesToAddress(b), err -} - -func decodeTopic(s string) (common.Hash, error) { - b, err := hexutil.Decode(s) - if err == nil && len(b) != common.HashLength { - err = fmt.Errorf("hex has invalid length %d after decoding", len(b)) - } - return common.BytesToHash(b), err -}