deep-stack/laconicd-deprecated
4
0
Fork 0
forked from cerc-io/laconicd-deprecated
Code Pull Requests Activity

rpc: event websocket subscription (#308)

Browse Source 
* rpc: event websocket subscription

* rpc: use tendermint event subscriptions

* new log events

* filter evm transactions

* filter logs

* wip: refactor filters

* remove custom BlockNumber

* wip: refactor rpc

* HeaderByNumber and HeaderByHash

* update Tendermint event system

* update Filter

* update EventSystem

* fix lint issues

* update rpc filters

* upgrade to tendermint v0.33.4

* update filters

* fix unsubscription

* updates wip

* initialize channels

* cleanup go routines

* pass ResultEvent channel on subscription

* error channel

* add block filter changes test

* add eventCh loop

* pass funcs in select go func, block filter working

* cleanup

* lint

* NewFilter and GetFilterChanges working

* eth_getLogs working

* lint

* lint

* cleanup

* remove logs and minor fixes

* changelog

* address @noot comments

* revert BlockNumber removal

Co-authored-by: noot <elizabethjbinks@gmail.com>
This commit is contained in:
Federico Kunze
2020-07-03 11:40:00 -04:00
committed by GitHub
co-authored by noot
parent c8b8f49675
commit 20e9b2ede3
12 changed files with 1304 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rpc/apis.go
+14 -9
View File
@@ -10,10 +10,15 @@ import (
"github.com/ethereum/go-ethereum/rpc"
)
const Web3Namespace = "web3"
const EthNamespace = "eth"
const PersonalNamespace = "personal"
const NetNamespace = "net"
// RPC namespaces and API version
const (
Web3Namespace = "web3"
EthNamespace = "eth"
PersonalNamespace = "personal"
NetNamespace = "net"
apiVersion = "1.0"
)
// GetRPCAPIs returns the list of all APIs
func GetRPCAPIs(cliCtx context.CLIContext, key emintcrypto.PrivKeySecp256k1) []rpc.API {
@@ -22,31 +27,31 @@ func GetRPCAPIs(cliCtx context.CLIContext, key emintcrypto.PrivKeySecp256k1) []r
return []rpc.API{
{
Namespace: Web3Namespace,
Version: "1.0",
Version: apiVersion,
Service: NewPublicWeb3API(),
Public: true,
},
{
Namespace: EthNamespace,
Version: "1.0",
Version: apiVersion,
Service: NewPublicEthAPI(cliCtx, backend, nonceLock, key),
Public: true,
},
{
Namespace: PersonalNamespace,
Version: "1.0",
Version: apiVersion,
Service: NewPersonalEthAPI(cliCtx, nonceLock),
Public: false,
},
{
Namespace: EthNamespace,
Version: "1.0",
Version: apiVersion,
Service: NewPublicFilterAPI(cliCtx, backend),
Public: true,
},
{
Namespace: NetNamespace,
Version: "1.0",
Version: apiVersion,
Service: NewPublicNetAPI(cliCtx),
Public: true,
},
rpc/backend.go
+128 -7
View File
@@ -5,6 +5,8 @@ import (
"math/big"
"strconv"
tmtypes "github.com/tendermint/tendermint/types"
evmtypes "github.com/cosmos/ethermint/x/evm/types"
"github.com/cosmos/cosmos-sdk/client/context"
@@ -19,20 +21,26 @@ import (
type Backend interface {
// Used by block filter; also used for polling
BlockNumber() (hexutil.Uint64, error)
HeaderByNumber(blockNum BlockNumber) (*ethtypes.Header, error)
HeaderByHash(blockHash common.Hash) (*ethtypes.Header, error)
GetBlockByNumber(blockNum BlockNumber, fullTx bool) (map[string]interface{}, error)
GetBlockByHash(hash common.Hash, fullTx bool) (map[string]interface{}, error)
getEthBlockByNumber(height int64, fullTx bool) (map[string]interface{}, error)
getGasLimit() (int64, error)
// returns the logs of a given block
GetLogs(blockHash common.Hash) ([][]*ethtypes.Log, error)
// Used by pending transaction filter
PendingTransactions() ([]*Transaction, error)
// Used by log filter
GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error)
// TODO: Bloom methods
BloomStatus() (uint64, uint64)
}
// EthermintBackend implements Backend
var _ Backend = (*EthermintBackend)(nil)
// EthermintBackend implements the Backend interface
type EthermintBackend struct {
cliCtx context.CLIContext
gasLimit int64
@@ -68,7 +76,7 @@ func (e *EthermintBackend) GetBlockByNumber(blockNum BlockNumber, fullTx bool) (
// GetBlockByHash returns the block identified by hash.
func (e *EthermintBackend) GetBlockByHash(hash common.Hash, fullTx bool) (map[string]interface{}, error) {
res, _, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryHashToHeight, hash.Hex()))
res, height, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryHashToHeight, hash.Hex()))
if err != nil {
return nil, err
}
@@ -78,9 +86,60 @@ func (e *EthermintBackend) GetBlockByHash(hash common.Hash, fullTx bool) (map[st
return nil, err
}
e.cliCtx = e.cliCtx.WithHeight(height)
return e.getEthBlockByNumber(out.Number, fullTx)
}
// HeaderByNumber returns the block header identified by height.
func (e *EthermintBackend) HeaderByNumber(blockNum BlockNumber) (*ethtypes.Header, error) {
return e.getBlockHeader(blockNum.Int64())
}
// HeaderByHash returns the block header identified by hash.
func (e *EthermintBackend) HeaderByHash(blockHash common.Hash) (*ethtypes.Header, error) {
res, height, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryHashToHeight, blockHash.Hex()))
if err != nil {
return nil, err
}
var out evmtypes.QueryResBlockNumber
if err := e.cliCtx.Codec.UnmarshalJSON(res, &out); err != nil {
return nil, err
}
e.cliCtx = e.cliCtx.WithHeight(height)
return e.getBlockHeader(out.Number)
}
func (e *EthermintBackend) getBlockHeader(height int64) (*ethtypes.Header, error) {
if height <= 0 {
// get latest block height
num, err := e.BlockNumber()
if err != nil {
return nil, err
}
height = int64(num)
}
block, err := e.cliCtx.Client.Block(&height)
if err != nil {
return nil, err
}
res, _, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryBloom, strconv.FormatInt(height, 10)))
if err != nil {
return nil, err
}
var bloomRes evmtypes.QueryBloomFilter
e.cliCtx.Codec.MustUnmarshalJSON(res, &bloomRes)
ethHeader := EthHeaderFromTendermint(block.Block.Header)
ethHeader.Bloom = bloomRes.Bloom
return ethHeader, nil
}
func (e *EthermintBackend) getEthBlockByNumber(height int64, fullTx bool) (map[string]interface{}, error) {
// Remove this check when 0 query is fixed ref: (https://github.com/tendermint/tendermint/issues/4014)
var blkNumPtr *int64
@@ -128,7 +187,6 @@ func (e *EthermintBackend) getEthBlockByNumber(height int64, fullTx bool) (map[s
var out evmtypes.QueryBloomFilter
e.cliCtx.Codec.MustUnmarshalJSON(res, &out)
return formatBlock(header, block.Block.Size(), gasLimit, gasUsed, transactions, out.Bloom), nil
}
@@ -158,11 +216,12 @@ func (e *EthermintBackend) getGasLimit() (int64, error) {
}
// GetTransactionLogs returns the logs given a transaction hash.
// It returns an error if there's an encoding error.
// If no logs are found for the tx hash, the error is nil.
func (e *EthermintBackend) GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error) {
// do we need to use the block height somewhere?
ctx := e.cliCtx
res, _, err := ctx.QueryWithData(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryTransactionLogs, txHash.Hex()), nil)
res, height, err := ctx.QueryWithData(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryTransactionLogs, txHash.Hex()), nil)
if err != nil {
return nil, err
}
@@ -172,6 +231,7 @@ func (e *EthermintBackend) GetTransactionLogs(txHash common.Hash) ([]*ethtypes.L
return nil, err
}
e.cliCtx = e.cliCtx.WithHeight(height)
return out.Logs, nil
}
@@ -183,7 +243,7 @@ func (e *EthermintBackend) PendingTransactions() ([]*Transaction, error) {
return nil, err
}
transactions := make([]*Transaction, 0, 100)
transactions := make([]*Transaction, pendingTxs.Count)
for _, tx := range pendingTxs.Txs {
ethTx, err := bytesToEthTx(e.cliCtx, tx)
if err != nil {
@@ -201,3 +261,64 @@ func (e *EthermintBackend) PendingTransactions() ([]*Transaction, error) {
return transactions, nil
}
// GetLogs returns all the logs from all the ethreum transactions in a block.
func (e *EthermintBackend) GetLogs(blockHash common.Hash) ([][]*ethtypes.Log, error) {
res, _, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryHashToHeight, blockHash.Hex()))
if err != nil {
return nil, err
}
var out evmtypes.QueryResBlockNumber
if err := e.cliCtx.Codec.UnmarshalJSON(res, &out); err != nil {
return nil, err
}
block, err := e.cliCtx.Client.Block(&out.Number)
if err != nil {
return nil, err
}
var blockLogs = [][]*ethtypes.Log{}
for _, tx := range block.Block.Txs {
// NOTE: we query the state in case the tx result logs are not persisted after an upgrade.
res, _, err := e.cliCtx.QueryWithData(fmt.Sprintf("custom/%s/%s/%s", evmtypes.ModuleName, evmtypes.QueryTransactionLogs, common.BytesToHash(tx.Hash()).Hex()), nil)
if err != nil {
continue
}
out := new(evmtypes.QueryETHLogs)
if err := e.cliCtx.Codec.UnmarshalJSON(res, &out); err != nil {
return nil, err
}
blockLogs = append(blockLogs, out.Logs)
}
return blockLogs, nil
}
// BloomStatus returns the BloomBitsBlocks and the number of processed sections maintained
// by the chain indexer.
func (e *EthermintBackend) BloomStatus() (uint64, uint64) {
return 4096, 0
}
// EthHeaderFromTendermint is an util function that returns an Ethereum Header
// from a tendermint Header.
func EthHeaderFromTendermint(header tmtypes.Header) *ethtypes.Header {
return &ethtypes.Header{
ParentHash: common.BytesToHash(header.LastBlockID.Hash.Bytes()),
UncleHash: common.Hash{},
Coinbase: common.Address{},
Root: common.BytesToHash(header.AppHash),
TxHash: common.BytesToHash(header.DataHash),
ReceiptHash: common.Hash{},
Difficulty: nil,
Number: big.NewInt(header.Height),
Time: uint64(header.Time.Unix()),
Extra: nil,
MixDigest: common.Hash{},
Nonce: ethtypes.BlockNonce{},
}
}
rpc/eth_api.go
+5 -5
View File
@@ -30,7 +30,6 @@ import (
"github.com/ethereum/go-ethereum/common/hexutil"
ethtypes "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/cosmos/cosmos-sdk/client/context"
"github.com/cosmos/cosmos-sdk/client/flags"
@@ -388,7 +387,7 @@ type CallArgs struct {
}
// Call performs a raw contract call.
func (e *PublicEthAPI) Call(args CallArgs, blockNr rpc.BlockNumber, overrides *map[common.Address]account) (hexutil.Bytes, error) {
func (e *PublicEthAPI) Call(args CallArgs, blockNr BlockNumber, overrides *map[common.Address]account) (hexutil.Bytes, error) {
simRes, err := e.doCall(args, blockNr, big.NewInt(emint.DefaultRPCGasLimit))
if err != nil {
return []byte{}, err
@@ -419,7 +418,7 @@ type account struct {
// DoCall performs a simulated call operation through the evmtypes. It returns the
// estimated gas used on the operation or an error if fails.
func (e *PublicEthAPI) doCall(
args CallArgs, blockNr rpc.BlockNumber, globalGasCap *big.Int,
args CallArgs, blockNr BlockNumber, globalGasCap *big.Int,
) (*sdk.SimulationResponse, error) {
// Set height for historical queries
ctx := e.cliCtx
@@ -561,7 +560,8 @@ func convertTransactionsToRPC(cliCtx context.CLIContext, txs []tmtypes.Tx, block
for i, tx := range txs {
ethTx, err := bytesToEthTx(cliCtx, tx)
if err != nil {
return nil, nil, err
// continue to next transaction in case it's not a MsgEthereumTx
continue
}
// TODO: Remove gas usage calculation if saving gasUsed per block
gasUsed.Add(gasUsed, ethTx.Fee())
@@ -602,7 +602,7 @@ func bytesToEthTx(cliCtx context.CLIContext, bz []byte) (*evmtypes.MsgEthereumTx
ethTx, ok := stdTx.(evmtypes.MsgEthereumTx)
if !ok {
return nil, fmt.Errorf("invalid transaction type, must be an amino encoded Ethereum transaction")
return nil, fmt.Errorf("invalid transaction type %T, expected MsgEthereumTx", stdTx)
}
return &ethTx, nil
}
rpc/filter_api.go
+527 -53
View File
@@ -1,79 +1,553 @@
package rpc
import (
"errors"
"context"
"fmt"
"sync"
"time"
"github.com/cosmos/cosmos-sdk/client/context"
coretypes "github.com/tendermint/tendermint/rpc/core/types"
tmtypes "github.com/tendermint/tendermint/types"
"github.com/ethereum/go-ethereum/common"
ethtypes "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/filters"
"github.com/ethereum/go-ethereum/rpc"
clientcontext "github.com/cosmos/cosmos-sdk/client/context"
evmtypes "github.com/cosmos/ethermint/x/evm/types"
)
// PublicFilterAPI is the eth_ prefixed set of APIs in the Web3 JSON-RPC spec.
type PublicFilterAPI struct {
cliCtx context.CLIContext
backend Backend
filters map[rpc.ID]*Filter // ID to filter; TODO: change to sync.Map in case of concurrent writes
// FiltersBackend defines the methods requided by the PublicFilterAPI backend
type FiltersBackend interface {
GetBlockByNumber(blockNum BlockNumber, fullTx bool) (map[string]interface{}, error)
HeaderByNumber(blockNr BlockNumber) (*ethtypes.Header, error)
HeaderByHash(blockHash common.Hash) (*ethtypes.Header, error)
GetLogs(blockHash common.Hash) ([][]*ethtypes.Log, error)
GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error)
BloomStatus() (uint64, uint64)
}
// NewPublicEthAPI creates an instance of the public ETH Web3 API.
func NewPublicFilterAPI(cliCtx context.CLIContext, backend Backend) *PublicFilterAPI {
return &PublicFilterAPI{
// consider a filter inactive if it has not been polled for within deadline
var deadline = 5 * time.Minute
// filter is a helper struct that holds meta information over the filter type
// and associated subscription in the event system.
type filter struct {
typ filters.Type
deadline *time.Timer // filter is inactive when deadline triggers
hashes []common.Hash
crit filters.FilterCriteria
logs []*ethtypes.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 as blocks, transactions and logs.
type PublicFilterAPI struct {
cliCtx clientcontext.CLIContext
backend FiltersBackend
events *EventSystem
filtersMu sync.Mutex
filters map[rpc.ID]*filter
}
// NewPublicFilterAPI returns a new PublicFilterAPI instance.
func NewPublicFilterAPI(cliCtx clientcontext.CLIContext, backend FiltersBackend) *PublicFilterAPI {
// start the client to subscribe to Tendermint events
err := cliCtx.Client.Start()
if err != nil {
panic(err)
}
api := &PublicFilterAPI{
cliCtx: cliCtx,
backend: backend,
filters: make(map[rpc.ID]*Filter),
filters: make(map[rpc.ID]*filter),
events: NewEventSystem(cliCtx.Client),
}
go api.timeoutLoop()
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(deadline)
defer ticker.Stop()
for {
<-ticker.C
api.filtersMu.Lock()
for id, f := range api.filters {
select {
case <-f.deadline.C:
f.s.Unsubscribe(api.events)
delete(api.filters, id)
default:
continue
}
}
api.filtersMu.Unlock()
}
}
// NewFilter instantiates a new filter.
func (e *PublicFilterAPI) NewFilter(criteria filters.FilterCriteria) rpc.ID {
id := rpc.NewID()
e.filters[id] = NewFilter(e.backend, &criteria)
return id
}
// NewBlockFilter instantiates a new block filter.
func (e *PublicFilterAPI) NewBlockFilter() rpc.ID {
id := rpc.NewID()
e.filters[id] = NewBlockFilter(e.backend)
return id
}
// NewPendingTransactionFilter instantiates a new pending transaction filter.
func (e *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
id := rpc.NewID()
e.filters[id] = NewPendingTransactionFilter(e.backend)
return id
}
// UninstallFilter uninstalls a filter with the given ID.
func (e *PublicFilterAPI) UninstallFilter(id rpc.ID) bool {
e.filters[id].uninstallFilter()
delete(e.filters, id)
return true
}
// GetFilterChanges returns an array of changes since the last poll.
// If the filter is a log filter, it returns an array of Logs.
// If the filter is a block filter, it returns an array of block hashes.
// If the filter is a pending transaction filter, it returns an array of transaction hashes.
func (e *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
if e.filters[id] == nil {
return nil, errors.New("invalid filter ID")
// 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 through 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 {
pendingTxSub, cancelSubs, err := api.events.SubscribePendingTxs()
if err != nil {
// wrap error on the ID
return rpc.ID(fmt.Sprintf("error creating pending tx filter: %s", err.Error()))
}
return e.filters[id].getFilterChanges()
api.filtersMu.Lock()
api.filters[pendingTxSub.ID()] = &filter{typ: filters.PendingTransactionsSubscription, deadline: time.NewTimer(deadline), hashes: make([]common.Hash, 0), s: pendingTxSub}
api.filtersMu.Unlock()
go func(txsCh <-chan coretypes.ResultEvent, errCh <-chan error) {
defer cancelSubs()
for {
select {
case ev := <-txsCh:
data, _ := ev.Data.(tmtypes.EventDataTx)
txHash := common.BytesToHash(data.Tx.Hash())
api.filtersMu.Lock()
if f, found := api.filters[pendingTxSub.ID()]; found {
f.hashes = append(f.hashes, txHash)
}
api.filtersMu.Unlock()
case <-errCh:
api.filtersMu.Lock()
delete(api.filters, pendingTxSub.ID())
api.filtersMu.Unlock()
}
}
}(pendingTxSub.eventCh, pendingTxSub.Err())
return pendingTxSub.ID()
}
// GetFilterLogs returns an array of all logs matching filter with given id.
func (e *PublicFilterAPI) GetFilterLogs(id rpc.ID) ([]*ethtypes.Log, error) {
return e.filters[id].getFilterLogs()
// 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()
ctx, cancelFn := context.WithTimeout(context.Background(), deadline)
defer cancelFn()
api.events.WithContext(ctx)
pendingTxSub, cancelSubs, err := api.events.SubscribePendingTxs()
if err != nil {
return nil, err
}
go func(txsCh <-chan coretypes.ResultEvent) {
defer cancelSubs()
for {
select {
case ev := <-txsCh:
data, _ := ev.Data.(tmtypes.EventDataTx)
txHash := common.BytesToHash(data.Tx.Hash())
// To keep the original behaviour, send a single tx hash in one notification.
// TODO(rjl493456442) Send a batch of tx hashes in one notification
err = notifier.Notify(rpcSub.ID, txHash)
if err != nil {
return
}
case <-rpcSub.Err():
pendingTxSub.Unsubscribe(api.events)
return
case <-notifier.Closed():
pendingTxSub.Unsubscribe(api.events)
return
}
}
}(pendingTxSub.eventCh)
return rpcSub, err
}
// 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 {
headerSub, cancelSubs, err := api.events.SubscribeNewHeads()
if err != nil {
// wrap error on the ID
return rpc.ID(fmt.Sprintf("error creating block filter: %s", err.Error()))
}
api.filtersMu.Lock()
api.filters[headerSub.ID()] = &filter{typ: filters.BlocksSubscription, deadline: time.NewTimer(deadline), hashes: []common.Hash{}, s: headerSub}
api.filtersMu.Unlock()
go func(headersCh <-chan coretypes.ResultEvent, errCh <-chan error) {
defer cancelSubs()
for {
select {
case ev := <-headersCh:
data, _ := ev.Data.(tmtypes.EventDataNewBlockHeader)
header := EthHeaderFromTendermint(data.Header)
api.filtersMu.Lock()
if f, found := api.filters[headerSub.ID()]; found {
f.hashes = append(f.hashes, header.Hash())
}
api.filtersMu.Unlock()
case <-errCh:
api.filtersMu.Lock()
delete(api.filters, headerSub.ID())
api.filtersMu.Unlock()
return
}
}
}(headerSub.eventCh, headerSub.Err())
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
}
api.events.WithContext(ctx)
rpcSub := notifier.CreateSubscription()
headersSub, cancelSubs, err := api.events.SubscribeNewHeads()
if err != nil {
return &rpc.Subscription{}, err
}
go func(headersCh <-chan coretypes.ResultEvent) {
defer cancelSubs()
for {
select {
case ev := <-headersCh:
data, ok := ev.Data.(tmtypes.EventDataNewBlockHeader)
if !ok {
err = fmt.Errorf("invalid event data %T, expected %s", ev.Data, tmtypes.EventNewBlockHeader)
headersSub.err <- err
return
}
header := EthHeaderFromTendermint(data.Header)
err = notifier.Notify(rpcSub.ID, header)
if err != nil {
headersSub.err <- err
return
}
case <-rpcSub.Err():
headersSub.Unsubscribe(api.events)
return
case <-notifier.Closed():
headersSub.Unsubscribe(api.events)
return
}
}
}(headersSub.eventCh)
return rpcSub, err
}
// Logs creates a subscription that fires for all new log that match the given filter criteria.
func (api *PublicFilterAPI) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
}
api.events.WithContext(ctx)
rpcSub := notifier.CreateSubscription()
logsSub, cancelSubs, err := api.events.SubscribeLogs(crit)
if err != nil {
return &rpc.Subscription{}, err
}
go func(logsCh <-chan coretypes.ResultEvent) {
defer cancelSubs()
for {
select {
case event := <-logsCh:
// filter only events from EVM module txs
_, isMsgEthermint := event.Events[evmtypes.TypeMsgEthermint]
_, isMsgEthereumTx := event.Events[evmtypes.TypeMsgEthereumTx]
if !(isMsgEthermint || isMsgEthereumTx) {
// ignore transaction as it's not from the evm module
return
}
// get transaction result data
dataTx, ok := event.Data.(tmtypes.EventDataTx)
if !ok {
err = fmt.Errorf("invalid event data %T, expected %s", event.Data, tmtypes.EventTx)
logsSub.err <- err
return
}
resultData, err := evmtypes.DecodeResultData(dataTx.TxResult.Result.Data)
if err != nil {
return
}
logs := filterLogs(resultData.Logs, crit.FromBlock, crit.ToBlock, crit.Addresses, crit.Topics)
for _, log := range logs {
err = notifier.Notify(rpcSub.ID, log)
if err != nil {
return
}
}
case <-rpcSub.Err(): // client send an unsubscribe request
logsSub.Unsubscribe(api.events)
return
case <-notifier.Closed(): // connection dropped
logsSub.Unsubscribe(api.events)
return
}
}
}(logsSub.eventCh)
return rpcSub, err
}
// 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(criteria filters.FilterCriteria) (rpc.ID, error) {
var (
filterID = rpc.ID("")
err error
)
logsSub, cancelSubs, err := api.events.SubscribeLogs(criteria)
if err != nil {
return rpc.ID(""), err
}
filterID = logsSub.ID()
api.filtersMu.Lock()
api.filters[filterID] = &filter{typ: filters.LogsSubscription, deadline: time.NewTimer(deadline), hashes: []common.Hash{}, s: logsSub}
api.filtersMu.Unlock()
go func(eventCh <-chan coretypes.ResultEvent) {
defer cancelSubs()
for {
select {
case event := <-eventCh:
dataTx, ok := event.Data.(tmtypes.EventDataTx)
if !ok {
err = fmt.Errorf("invalid event data %T, expected EventDataTx", event.Data)
return
}
var resultData evmtypes.ResultData
resultData, err = evmtypes.DecodeResultData(dataTx.TxResult.Result.Data)
if err != nil {
return
}
logs := filterLogs(resultData.Logs, criteria.FromBlock, criteria.ToBlock, criteria.Addresses, criteria.Topics)
api.filtersMu.Lock()
if f, found := api.filters[filterID]; found {
f.logs = append(f.logs, logs...)
}
api.filtersMu.Unlock()
case <-logsSub.Err():
api.filtersMu.Lock()
delete(api.filters, filterID)
api.filtersMu.Unlock()
return
}
}
}(logsSub.eventCh)
return filterID, err
}
// GetLogs returns logs matching the given argument that are stored within the state.
//
// https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs
func (e *PublicFilterAPI) GetLogs(criteria filters.FilterCriteria) ([]*ethtypes.Log, error) {
filter := NewFilter(e.backend, &criteria)
return filter.getFilterLogs()
func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit filters.FilterCriteria) ([]*ethtypes.Log, error) {
var filter *Filter
if crit.BlockHash != nil {
// Block filter requested, construct a single-shot filter
filter = NewBlockFilter(api.backend, crit)
} else {
// Convert the RPC block numbers into internal representations
begin := rpc.LatestBlockNumber.Int64()
if crit.FromBlock != nil {
begin = crit.FromBlock.Int64()
}
end := rpc.LatestBlockNumber.Int64()
if crit.ToBlock != nil {
end = crit.ToBlock.Int64()
}
// Construct the range filter
filter = NewRangeFilter(api.backend, begin, end, crit.Addresses, crit.Topics)
}
// Run the filter and return all the logs
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 {
return false
}
f.s.Unsubscribe(api.events)
return true
}
// 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) ([]*ethtypes.Log, error) {
api.filtersMu.Lock()
f, found := api.filters[id]
api.filtersMu.Unlock()
if !found {
return returnLogs(nil), fmt.Errorf("filter %s not found", id)
}
if f.typ != filters.LogsSubscription {
return returnLogs(nil), fmt.Errorf("filter %s doesn't have a LogsSubscription type: got %d", id, f.typ)
}
var filter *Filter
if f.crit.BlockHash != nil {
// Block filter requested, construct a single-shot filter
filter = NewBlockFilter(api.backend, f.crit)
} else {
// Convert the RPC block numbers into internal representations
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()
}
// Construct the range filter
filter = NewRangeFilter(api.backend, begin, end, f.crit.Addresses, f.crit.Topics)
}
// Run the filter and return all the logs
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()
f, found := api.filters[id]
if !found {
return nil, fmt.Errorf("filter %s not found", id)
}
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 filters.PendingTransactionsSubscription, filters.BlocksSubscription:
hashes := f.hashes
f.hashes = nil
return returnHashes(hashes), nil
case filters.LogsSubscription, filters.MinedAndPendingLogsSubscription:
logs := make([]*ethtypes.Log, len(f.logs))
copy(logs, f.logs)
f.logs = []*ethtypes.Log{}
return returnLogs(logs), nil
default:
return nil, fmt.Errorf("invalid filter %s type %d", id, f.typ)
}
}
// 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
}
// returnLogs is a helper that will return an empty log array in case the given logs array is nil,
// otherwise the given logs array is returned.
func returnLogs(logs []*ethtypes.Log) []*ethtypes.Log {
if logs == nil {
return []*ethtypes.Log{}
}
return logs
}
rpc/filter_system.go
+433
View File
@@ -0,0 +1,433 @@
package rpc
import (
"context"
"fmt"
"log"
"time"
tmquery "github.com/tendermint/tendermint/libs/pubsub/query"
rpcclient "github.com/tendermint/tendermint/rpc/client"
coretypes "github.com/tendermint/tendermint/rpc/core/types"
tmtypes "github.com/tendermint/tendermint/types"
"github.com/ethereum/go-ethereum/common"
ethtypes "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/filters"
"github.com/ethereum/go-ethereum/rpc"
sdk "github.com/cosmos/cosmos-sdk/types"
evmtypes "github.com/cosmos/ethermint/x/evm/types"
)
var (
txEvents = tmtypes.QueryForEvent(tmtypes.EventTx).String()
evmEvents = tmquery.MustParse(fmt.Sprintf("%s='%s' AND %s.%s='%s'", tmtypes.EventTypeKey, tmtypes.EventTx, sdk.EventTypeMessage, sdk.AttributeKeyModule, evmtypes.ModuleName)).String()
headerEvents = tmtypes.QueryForEvent(tmtypes.EventNewBlockHeader).String()
)
// EventSystem creates subscriptions, processes events and broadcasts them to the
// subscription which match the subscription criteria using the Tendermint's RPC client.
type EventSystem struct {
ctx context.Context
client rpcclient.Client
// light client mode
lightMode bool
index filterIndex
// Subscriptions
txsSub *Subscription // Subscription for new transaction event
logsSub *Subscription // Subscription for new log event
// rmLogsSub *Subscription // Subscription for removed log event
pendingLogsSub *Subscription // Subscription for pending log event
chainSub *Subscription // Subscription for new chain event
// Channels
install chan *Subscription // install filter for event notification
uninstall chan *Subscription // remove filter for event notification
// Unidirectional channels to receive Tendermint ResultEvents
txsCh <-chan coretypes.ResultEvent // Channel to receive new pending transactions event
logsCh <-chan coretypes.ResultEvent // Channel to receive new log event
pendingLogsCh <-chan coretypes.ResultEvent // Channel to receive new pending log event
// rmLogsCh <-chan coretypes.ResultEvent // Channel to receive removed log event
chainCh <-chan coretypes.ResultEvent // Channel to receive new chain event
}
// NewEventSystem 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 NewEventSystem(client rpcclient.Client) *EventSystem {
index := make(filterIndex)
for i := filters.UnknownSubscription; i < filters.LastIndexSubscription; i++ {
index[i] = make(map[rpc.ID]*Subscription)
}
es := &EventSystem{
ctx: context.Background(),
client: client,
lightMode: false,
index: index,
install: make(chan *Subscription),
uninstall: make(chan *Subscription),
txsCh: make(<-chan coretypes.ResultEvent),
logsCh: make(<-chan coretypes.ResultEvent),
pendingLogsCh: make(<-chan coretypes.ResultEvent),
// rmLogsCh: make(<-chan coretypes.ResultEvent),
chainCh: make(<-chan coretypes.ResultEvent),
}
go es.eventLoop()
return es
}
// WithContext sets a new context to the EventSystem. This is required to set a timeout context when
// a new filter is intantiated.
func (es *EventSystem) WithContext(ctx context.Context) {
es.ctx = ctx
}
// subscribe performs a new event subscription to a given Tendermint event.
// The subscription creates a unidirectional receive event channel to receive the ResultEvent. By
// default, the subscription timeouts (i.e is canceled) after 5 minutes. This function returns an
// error if the subscription fails (eg: if the identifier is already subscribed) or if the filter
// type is invalid.
func (es *EventSystem) subscribe(sub *Subscription) (*Subscription, context.CancelFunc, error) {
var (
err error
cancelFn context.CancelFunc
eventCh <-chan coretypes.ResultEvent
)
es.ctx, cancelFn = context.WithTimeout(context.Background(), deadline)
switch sub.typ {
case filters.PendingTransactionsSubscription:
eventCh, err = es.client.Subscribe(es.ctx, string(sub.id), sub.event)
case filters.PendingLogsSubscription, filters.MinedAndPendingLogsSubscription:
eventCh, err = es.client.Subscribe(es.ctx, string(sub.id), sub.event)
case filters.LogsSubscription:
eventCh, err = es.client.Subscribe(es.ctx, string(sub.id), sub.event)
case filters.BlocksSubscription:
eventCh, err = es.client.Subscribe(es.ctx, string(sub.id), sub.event)
default:
err = fmt.Errorf("invalid filter subscription type %d", sub.typ)
}
if err != nil {
sub.err <- err
return nil, cancelFn, err
}
// wrap events in a go routine to prevent blocking
go func() {
es.install <- sub
<-sub.installed
}()
sub.eventCh = eventCh
return sub, cancelFn, nil
}
// SubscribeLogs creates a subscription that will write all logs matching the
// given criteria to the given logs channel. Default value for the from and to
// block is "latest". If the fromBlock > toBlock an error is returned.
func (es *EventSystem) SubscribeLogs(crit filters.FilterCriteria) (*Subscription, context.CancelFunc, error) {
var from, to rpc.BlockNumber
if crit.FromBlock == nil {
from = rpc.LatestBlockNumber
} else {
from = rpc.BlockNumber(crit.FromBlock.Int64())
}
if crit.ToBlock == nil {
to = rpc.LatestBlockNumber
} else {
to = rpc.BlockNumber(crit.ToBlock.Int64())
}
switch {
// only interested in pending logs
case from == rpc.PendingBlockNumber && to == rpc.PendingBlockNumber:
return es.subscribePendingLogs(crit)
// only interested in new mined logs, mined logs within a specific block range, or
// logs from a specific block number to new mined blocks
case (from == rpc.LatestBlockNumber && to == rpc.LatestBlockNumber),
(from >= 0 && to >= 0 && to >= from):
return es.subscribeLogs(crit)
// interested in mined logs from a specific block number, new logs and pending logs
case from >= rpc.LatestBlockNumber && (to == rpc.PendingBlockNumber || to == rpc.LatestBlockNumber):
return es.subscribeMinedPendingLogs(crit)
default:
return nil, nil, fmt.Errorf("invalid from and to block combination: from > to (%d > %d)", from, to)
}
}
// subscribeMinedPendingLogs creates a subscription that returned mined and
// pending logs that match the given criteria.
func (es *EventSystem) subscribeMinedPendingLogs(crit filters.FilterCriteria) (*Subscription, context.CancelFunc, error) {
sub := &Subscription{
id: rpc.NewID(),
typ: filters.MinedAndPendingLogsSubscription,
event: evmEvents,
logsCrit: crit,
created: time.Now().UTC(),
logs: make(chan []*ethtypes.Log),
installed: make(chan struct{}, 1),
err: make(chan error, 1),
}
return es.subscribe(sub)
}
// subscribeLogs creates a subscription that will write all logs matching the
// given criteria to the given logs channel.
func (es *EventSystem) subscribeLogs(crit filters.FilterCriteria) (*Subscription, context.CancelFunc, error) {
sub := &Subscription{
id: rpc.NewID(),
typ: filters.LogsSubscription,
event: evmEvents,
logsCrit: crit,
created: time.Now().UTC(),
logs: make(chan []*ethtypes.Log),
installed: make(chan struct{}, 1),
err: make(chan error, 1),
}
return es.subscribe(sub)
}
// subscribePendingLogs creates a subscription that writes transaction hashes for
// transactions that enter the transaction pool.
func (es *EventSystem) subscribePendingLogs(crit filters.FilterCriteria) (*Subscription, context.CancelFunc, error) {
sub := &Subscription{
id: rpc.NewID(),
typ: filters.PendingLogsSubscription,
event: evmEvents,
logsCrit: crit,
created: time.Now().UTC(),
logs: make(chan []*ethtypes.Log),
installed: make(chan struct{}, 1),
err: make(chan error, 1),
}
return es.subscribe(sub)
}
// SubscribeNewHeads subscribes to new block headers events.
func (es EventSystem) SubscribeNewHeads() (*Subscription, context.CancelFunc, error) {
sub := &Subscription{
id: rpc.NewID(),
typ: filters.BlocksSubscription,
event: headerEvents,
created: time.Now().UTC(),
headers: make(chan *ethtypes.Header),
installed: make(chan struct{}, 1),
err: make(chan error, 1),
}
return es.subscribe(sub)
}
// SubscribePendingTxs subscribes to new pending transactions events from the mempool.
func (es EventSystem) SubscribePendingTxs() (*Subscription, context.CancelFunc, error) {
sub := &Subscription{
id: rpc.NewID(),
typ: filters.PendingTransactionsSubscription,
event: txEvents,
created: time.Now().UTC(),
hashes: make(chan []common.Hash),
installed: make(chan struct{}, 1),
err: make(chan error, 1),
}
return es.subscribe(sub)
}
type filterIndex map[filters.Type]map[rpc.ID]*Subscription
func (es *EventSystem) handleLogs(ev coretypes.ResultEvent) {
data, _ := ev.Data.(tmtypes.EventDataTx)
resultData, err := evmtypes.DecodeResultData(data.TxResult.Result.Data)
if err != nil {
return
}
if len(resultData.Logs) == 0 {
return
}
for _, f := range es.index[filters.LogsSubscription] {
matchedLogs := filterLogs(resultData.Logs, f.logsCrit.FromBlock, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics)
if len(matchedLogs) > 0 {
f.logs <- matchedLogs
}
}
}
func (es *EventSystem) handleTxsEvent(ev coretypes.ResultEvent) {
data, _ := ev.Data.(tmtypes.EventDataTx)
for _, f := range es.index[filters.PendingTransactionsSubscription] {
f.hashes <- []common.Hash{common.BytesToHash(data.Tx.Hash())}
}
}
func (es *EventSystem) handleChainEvent(ev coretypes.ResultEvent) {
data, _ := ev.Data.(tmtypes.EventDataNewBlockHeader)
for _, f := range es.index[filters.BlocksSubscription] {
f.headers <- EthHeaderFromTendermint(data.Header)
}
// TODO: light client
}
// eventLoop (un)installs filters and processes mux events.
func (es *EventSystem) eventLoop() {
var (
err error
cancelPendingTxsSubs, cancelLogsSubs, cancelPendingLogsSubs, cancelHeaderSubs context.CancelFunc
)
// Subscribe events
es.txsSub, cancelPendingTxsSubs, err = es.SubscribePendingTxs()
if err != nil {
panic(fmt.Errorf("failed to subscribe pending txs: %w", err))
}
defer cancelPendingTxsSubs()
es.logsSub, cancelLogsSubs, err = es.SubscribeLogs(filters.FilterCriteria{})
if err != nil {
panic(fmt.Errorf("failed to subscribe logs: %w", err))
}
defer cancelLogsSubs()
es.pendingLogsSub, cancelPendingLogsSubs, err = es.subscribePendingLogs(filters.FilterCriteria{})
if err != nil {
panic(fmt.Errorf("failed to subscribe pending logs: %w", err))
}
defer cancelPendingLogsSubs()
es.chainSub, cancelHeaderSubs, err = es.SubscribeNewHeads()
if err != nil {
panic(fmt.Errorf("failed to subscribe headers: %w", err))
}
defer cancelHeaderSubs()
// Ensure all subscriptions get cleaned up
defer func() {
es.txsSub.Unsubscribe(es)
es.logsSub.Unsubscribe(es)
// es.rmLogsSub.Unsubscribe(es)
es.pendingLogsSub.Unsubscribe(es)
es.chainSub.Unsubscribe(es)
}()
for {
select {
case txEvent := <-es.txsSub.eventCh:
es.handleTxsEvent(txEvent)
case headerEv := <-es.chainSub.eventCh:
es.handleChainEvent(headerEv)
case logsEv := <-es.logsSub.eventCh:
es.handleLogs(logsEv)
// TODO: figure out how to handle removed logs
// case logsEv := <-es.rmLogsSub.eventCh:
// es.handleLogs(logsEv)
case logsEv := <-es.pendingLogsSub.eventCh:
es.handleLogs(logsEv)
case f := <-es.install:
if f.typ == filters.MinedAndPendingLogsSubscription {
// the type are logs and pending logs subscriptions
es.index[filters.LogsSubscription][f.id] = f
es.index[filters.PendingLogsSubscription][f.id] = f
} else {
es.index[f.typ][f.id] = f
}
close(f.installed)
case f := <-es.uninstall:
if f.typ == filters.MinedAndPendingLogsSubscription {
// the type are logs and pending logs subscriptions
delete(es.index[filters.LogsSubscription], f.id)
delete(es.index[filters.PendingLogsSubscription], f.id)
} else {
delete(es.index[f.typ], f.id)
}
close(f.err)
// System stopped
case <-es.txsSub.Err():
return
case <-es.logsSub.Err():
return
// case <-es.rmLogsSub.Err():
// return
case <-es.pendingLogsSub.Err():
return
case <-es.chainSub.Err():
return
}
}
// }()
}
// Subscription defines a wrapper for the private subscription
type Subscription struct {
id rpc.ID
typ filters.Type
event string
created time.Time
logsCrit filters.FilterCriteria
logs chan []*ethtypes.Log
hashes chan []common.Hash
headers chan *ethtypes.Header
installed chan struct{} // closed when the filter is installed
eventCh <-chan coretypes.ResultEvent
err chan error
}
// ID returns the underlying subscription RPC identifier.
func (s Subscription) ID() rpc.ID {
return s.id
}
// Unsubscribe to the current subscription from Tendermint Websocket. It sends an error to the
// subscription error channel if unsubscription fails.
func (s *Subscription) Unsubscribe(es *EventSystem) {
if err := es.client.Unsubscribe(es.ctx, string(s.ID()), s.event); err != nil {
s.err <- err
}
go func() {
defer func() {
log.Println("successfully unsubscribed to event", s.event)
}()
uninstallLoop:
for {
// write uninstall request and consume logs/hashes. This prevents
// the eventLoop broadcast method to deadlock when writing to the
// filter event channel while the subscription loop is waiting for
// this method to return (and thus not reading these events).
select {
case es.uninstall <- s:
break uninstallLoop
case <-s.logs:
case <-s.hashes:
case <-s.headers:
}
}
}()
}
// Err returns the error channel
func (s *Subscription) Err() <-chan error {
return s.err
}
rpc/filters.go
+137 -230
View File
@@ -1,287 +1,168 @@
package rpc
import (
"errors"
"context"
"fmt"
"math/big"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/bloombits"
ethtypes "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/filters"
"github.com/ethereum/go-ethereum/log"
)
/*
- Filter functions derived from go-ethereum
Used to set the criteria passed in from RPC params
*/
const blockFilter = "block"
const pendingTxFilter = "pending"
const logFilter = "log"
// Filter can be used to retrieve and filter logs, blocks, or pending transactions.
// Filter can be used to retrieve and filter logs.
type Filter struct {
backend Backend
fromBlock, toBlock *big.Int // start and end block numbers
addresses []common.Address // contract addresses to watch
topics [][]common.Hash // log topics to watch for
blockHash *common.Hash // Block hash if filtering a single block
typ string
hashes []common.Hash // filtered block or transaction hashes
logs []*ethtypes.Log //nolint // filtered logs
stopped bool // set to true once filter in uninstalled
err error
backend FiltersBackend
criteria filters.FilterCriteria
matcher *bloombits.Matcher
}
// NewFilter returns a new Filter
func NewFilter(backend Backend, criteria *filters.FilterCriteria) *Filter {
filter := &Filter{
backend: backend,
fromBlock: criteria.FromBlock,
toBlock: criteria.ToBlock,
addresses: criteria.Addresses,
topics: criteria.Topics,
typ: logFilter,
stopped: false,
// NewBlockFilter creates a new filter which directly inspects the contents of
// a block to figure out whether it is interesting or not.
func NewBlockFilter(backend FiltersBackend, criteria filters.FilterCriteria) *Filter {
// Create a generic filter and convert it into a block filter
return newFilter(backend, criteria, nil)
}
// NewRangeFilter creates a new filter which uses a bloom filter on blocks to
// figure out whether a particular block is interesting or not.
func NewRangeFilter(backend FiltersBackend, begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter {
// Flatten the address and topic filter clauses into a single bloombits filter
// system. Since the bloombits are not positional, nil topics are permitted,
// which get flattened into a nil byte slice.
var filtersBz [][][]byte // nolint: prealloc
if len(addresses) > 0 {
filter := make([][]byte, len(addresses))
for i, address := range addresses {
filter[i] = address.Bytes()
}
filtersBz = append(filtersBz, filter)
}
return filter
for _, topicList := range topics {
filter := make([][]byte, len(topicList))
for i, topic := range topicList {
filter[i] = topic.Bytes()
}
filtersBz = append(filtersBz, filter)
}
size, _ := backend.BloomStatus()
// Create a generic filter and convert it into a range filter
criteria := filters.FilterCriteria{
FromBlock: big.NewInt(begin),
ToBlock: big.NewInt(end),
Addresses: addresses,
Topics: topics,
}
return newFilter(backend, criteria, bloombits.NewMatcher(size, filtersBz))
}
// NewFilterWithBlockHash returns a new Filter with a blockHash.
func NewFilterWithBlockHash(backend Backend, criteria *filters.FilterCriteria) *Filter {
// newFilter returns a new Filter
func newFilter(backend FiltersBackend, criteria filters.FilterCriteria, matcher *bloombits.Matcher) *Filter {
return &Filter{
backend: backend,
fromBlock: criteria.FromBlock,
toBlock: criteria.ToBlock,
addresses: criteria.Addresses,
topics: criteria.Topics,
blockHash: criteria.BlockHash,
typ: logFilter,
backend: backend,
criteria: criteria,
matcher: matcher,
}
}
// NewBlockFilter creates a new filter that notifies when a block arrives.
func NewBlockFilter(backend Backend) *Filter {
filter := NewFilter(backend, &filters.FilterCriteria{})
filter.typ = blockFilter
// Logs searches the blockchain for matching log entries, returning all from the
// first block that contains matches, updating the start of the filter accordingly.
func (f *Filter) Logs(_ context.Context) ([]*ethtypes.Log, error) {
logs := []*ethtypes.Log{}
var err error
go func() {
err := filter.pollForBlocks()
if err != nil {
filter.err = err
}
}()
return filter
}
func (f *Filter) pollForBlocks() error {
prev := hexutil.Uint64(0)
for {
if f.stopped {
return nil
}
num, err := f.backend.BlockNumber()
if err != nil {
return err
}
if num == prev {
continue
}
block, err := f.backend.GetBlockByNumber(BlockNumber(num), false)
if err != nil {
return err
}
hashBytes, ok := block["hash"].(hexutil.Bytes)
if !ok {
return errors.New("could not convert block hash to hexutil.Bytes")
}
hash := common.BytesToHash(hashBytes)
f.hashes = append(f.hashes, hash)
prev = num
// TODO: should we add a delay?
}
}
func (f *Filter) pollForTransactions() error {
for {
if f.stopped {
return nil
}
txs, err := f.backend.PendingTransactions()
if err != nil {
return err
}
for _, tx := range txs {
if !contains(f.hashes, tx.Hash) {
f.hashes = append(f.hashes, tx.Hash)
}
}
<-time.After(1 * time.Second)
}
}
func contains(slice []common.Hash, item common.Hash) bool {
set := make(map[common.Hash]struct{}, len(slice))
for _, s := range slice {
set[s] = struct{}{}
}
_, ok := set[item]
return ok
}
// NewPendingTransactionFilter creates a new filter that notifies when a pending transaction arrives.
func NewPendingTransactionFilter(backend Backend) *Filter {
filter := NewFilter(backend, &filters.FilterCriteria{})
filter.typ = pendingTxFilter
go func() {
err := filter.pollForTransactions()
if err != nil {
filter.err = err
}
}()
return filter
}
func (f *Filter) uninstallFilter() {
f.stopped = true
}
func (f *Filter) getFilterChanges() (interface{}, error) {
switch f.typ {
case blockFilter:
if f.err != nil {
return nil, f.err
}
blocks := make([]common.Hash, len(f.hashes))
copy(blocks, f.hashes)
f.hashes = []common.Hash{}
return blocks, nil
case pendingTxFilter:
if f.err != nil {
return nil, f.err
}
txs := make([]common.Hash, len(f.hashes))
copy(txs, f.hashes)
f.hashes = []common.Hash{}
return txs, nil
case logFilter:
return f.getFilterLogs()
}
return nil, errors.New("unsupported filter")
}
func (f *Filter) getFilterLogs() ([]*ethtypes.Log, error) {
ret := []*ethtypes.Log{}
// filter specific block only
if f.blockHash != nil {
block, err := f.backend.GetBlockByHash(*f.blockHash, true)
// If we're doing singleton block filtering, execute and return
if f.criteria.BlockHash != nil && f.criteria.BlockHash != (&common.Hash{}) {
header, err := f.backend.HeaderByHash(*f.criteria.BlockHash)
if err != nil {
return nil, err
}
// if the logsBloom == 0, there are no logs in that block
if txs, ok := block["transactions"].([]common.Hash); !ok {
return ret, nil
} else if len(txs) != 0 {
return f.checkMatches(block)
if header == nil {
return nil, fmt.Errorf("unknown block header %s", f.criteria.BlockHash.String())
}
return f.blockLogs(header)
}
// filter range of blocks
num, err := f.backend.BlockNumber()
// Figure out the limits of the filter range
header, err := f.backend.HeaderByNumber(LatestBlockNumber)
if err != nil {
return nil, err
}
// if f.fromBlock is set to 0, set it to the latest block number
if f.fromBlock == nil || f.fromBlock.Cmp(big.NewInt(0)) == 0 {
f.fromBlock = big.NewInt(int64(num))
if header == nil || header.Number == nil {
return nil, nil
}
// if f.toBlock is set to 0, set it to the latest block number
if f.toBlock == nil || f.toBlock.Cmp(big.NewInt(0)) == 0 {
f.toBlock = big.NewInt(int64(num))
head := header.Number.Int64()
if f.criteria.FromBlock.Int64() == -1 {
f.criteria.FromBlock = big.NewInt(head)
}
if f.criteria.ToBlock.Int64() == -1 {
f.criteria.ToBlock = big.NewInt(head)
}
log.Debug("[ethAPI] Retrieving filter logs", "fromBlock", f.fromBlock, "toBlock", f.toBlock,
"topics", f.topics, "addresses", f.addresses)
from := f.fromBlock.Int64()
to := f.toBlock.Int64()
for i := from; i <= to; i++ {
block, err := f.backend.GetBlockByNumber(NewBlockNumber(big.NewInt(i)), true)
for i := f.criteria.FromBlock.Int64(); i <= f.criteria.ToBlock.Int64(); i++ {
block, err := f.backend.GetBlockByNumber(BlockNumber(i), true)
if err != nil {
f.err = err
log.Debug("[ethAPI] Cannot get block", "block", block["number"], "error", err)
break
return logs, err
}
log.Debug("[ethAPI] filtering", "block", block)
// TODO: block logsBloom is often set in the wrong block
// if the logsBloom == 0, there are no logs in that block
if txs, ok := block["transactions"].([]common.Hash); !ok {
txs, ok := block["transactions"].([]common.Hash)
if !ok || len(txs) == 0 {
continue
} else if len(txs) != 0 {
logs, err := f.checkMatches(block)
if err != nil {
f.err = err
break
}
ret = append(ret, logs...)
}
logsMatched := f.checkMatches(txs)
logs = append(logs, logsMatched...)
}
return ret, nil
return logs, nil
}
func (f *Filter) checkMatches(block map[string]interface{}) ([]*ethtypes.Log, error) {
transactions, ok := block["transactions"].([]common.Hash)
if !ok {
return nil, errors.New("invalid block transactions")
// blockLogs returns the logs matching the filter criteria within a single block.
func (f *Filter) blockLogs(header *ethtypes.Header) ([]*ethtypes.Log, error) {
if !bloomFilter(header.Bloom, f.criteria.Addresses, f.criteria.Topics) {
return []*ethtypes.Log{}, nil
}
unfiltered := []*ethtypes.Log{}
logsList, err := f.backend.GetLogs(header.Hash())
if err != nil {
return []*ethtypes.Log{}, err
}
var unfiltered []*ethtypes.Log // nolint: prealloc
for _, logs := range logsList {
unfiltered = append(unfiltered, logs...)
}
logs := filterLogs(unfiltered, nil, nil, f.criteria.Addresses, f.criteria.Topics)
if len(logs) == 0 {
return []*ethtypes.Log{}, nil
}
return logs, nil
}
// checkMatches checks if the logs from the a list of transactions transaction
// contain any log events that match the filter criteria. This function is
// called when the bloom filter signals a potential match.
func (f *Filter) checkMatches(transactions []common.Hash) []*ethtypes.Log {
unfiltered := []*ethtypes.Log{}
for _, tx := range transactions {
logs, err := f.backend.GetTransactionLogs(common.BytesToHash(tx[:]))
logs, err := f.backend.GetTransactionLogs(tx)
if err != nil {
return nil, err
// ignore error if transaction didn't set any logs (eg: when tx type is not
// MsgEthereumTx or MsgEthermint)
continue
}
unfiltered = append(unfiltered, logs...)
}
return filterLogs(unfiltered, f.fromBlock, f.toBlock, f.addresses, f.topics), nil
return filterLogs(unfiltered, f.criteria.FromBlock, f.criteria.ToBlock, f.criteria.Addresses, f.criteria.Topics)
}
// filterLogs creates a slice of logs matching the given criteria.
@@ -305,7 +186,7 @@ Logs:
}
// If the to filtered topics is greater than the amount of topics in logs, skip.
if len(topics) > len(log.Topics) {
continue Logs
continue
}
for i, sub := range topics {
match := len(sub) == 0 // empty rule set == wildcard
@@ -333,3 +214,29 @@ func includes(addresses []common.Address, a common.Address) bool {
return false
}
func bloomFilter(bloom ethtypes.Bloom, addresses []common.Address, topics [][]common.Hash) bool {
var included bool
if len(addresses) > 0 {
for _, addr := range addresses {
if ethtypes.BloomLookup(bloom, addr) {
included = true
break
}
}
if !included {
return false
}
}
for _, sub := range topics {
included = len(sub) == 0 // empty rule set == wildcard
for _, topic := range sub {
if ethtypes.BloomLookup(bloom, topic) {
included = true
break
}
}
}
return included
}
rpc/net_api.go
+2 -2
View File
@@ -15,8 +15,8 @@ type PublicNetAPI struct {
networkVersion uint64
}
// NewPersonalEthAPI creates an instance of the public ETH Web3 API.
func NewPublicNetAPI(cliCtx context.CLIContext) *PublicNetAPI {
// NewPublicNetAPI creates an instance of the public Net Web3 API.
func NewPublicNetAPI(_ context.CLIContext) *PublicNetAPI {
chainID := viper.GetString(flags.FlagChainID)
// parse the chainID from a integer string
intChainID, err := strconv.ParseUint(chainID, 0, 64)
rpc/types.go
+1
View File
@@ -20,6 +20,7 @@ const (
EarliestBlockNumber = BlockNumber(1)
)
// NewBlockNumber creates a new BlockNumber instance.
func NewBlockNumber(n *big.Int) BlockNumber {
return BlockNumber(n.Int64())
}