mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-18 18:02:24 +00:00
eth/filters,rpc,core,internal/ethapi: Fixed formating, linting, vet issues.
This commit is contained in:
parent
23f0df9db6
commit
169c590b07
15 changed files with 151 additions and 138 deletions
|
|
@ -461,6 +461,10 @@ func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscr
|
||||||
return fb.bc.SubscribeLogsEvent(ch)
|
return fb.bc.SubscribeLogsEvent(ch)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (fb *filterBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription {
|
||||||
|
return fb.bc.SubscribeTransactionEvent(ch)
|
||||||
|
}
|
||||||
|
|
||||||
func (fb *filterBackend) BloomStatus() (uint64, uint64) { return 4096, 0 }
|
func (fb *filterBackend) BloomStatus() (uint64, uint64) { return 4096, 0 }
|
||||||
func (fb *filterBackend) ServiceFilter(ctx context.Context, ms *bloombits.MatcherSession) {
|
func (fb *filterBackend) ServiceFilter(ctx context.Context, ms *bloombits.MatcherSession) {
|
||||||
panic("not supported")
|
panic("not supported")
|
||||||
|
|
|
||||||
|
|
@ -99,7 +99,7 @@ type BlockChain struct {
|
||||||
chainSideFeed event.Feed
|
chainSideFeed event.Feed
|
||||||
chainHeadFeed event.Feed
|
chainHeadFeed event.Feed
|
||||||
logsFeed event.Feed
|
logsFeed event.Feed
|
||||||
txPostFeed event.Feed // named with Post to distinguish between txFeed in tx_pool, which gets triggered *before* a transaction gets processed rather than after
|
txPostFeed event.Feed // named with Post to distinguish between txFeed in tx_pool, which gets triggered *before* a transaction gets processed rather than after
|
||||||
scope event.SubscriptionScope
|
scope event.SubscriptionScope
|
||||||
genesisBlock *types.Block
|
genesisBlock *types.Block
|
||||||
|
|
||||||
|
|
@ -1372,8 +1372,8 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) {
|
||||||
|
|
||||||
case ChainSideEvent:
|
case ChainSideEvent:
|
||||||
bc.chainSideFeed.Send(ev)
|
bc.chainSideFeed.Send(ev)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (bc *BlockChain) update() {
|
func (bc *BlockChain) update() {
|
||||||
|
|
|
||||||
|
|
@ -36,9 +36,9 @@ type PendingStateEvent struct{}
|
||||||
type NewMinedBlockEvent struct{ Block *types.Block }
|
type NewMinedBlockEvent struct{ Block *types.Block }
|
||||||
|
|
||||||
// TransactionEvent is posted when a transaction completes execution
|
// TransactionEvent is posted when a transaction completes execution
|
||||||
type TransactionEvent struct{
|
type TransactionEvent struct {
|
||||||
TxHash common.Hash
|
TxHash common.Hash
|
||||||
RetData types.ReturnData
|
RetData types.ReturnData
|
||||||
}
|
}
|
||||||
|
|
||||||
// RemovedTransactionEvent is posted when a reorg happens
|
// RemovedTransactionEvent is posted when a reorg happens
|
||||||
|
|
|
||||||
|
|
@ -124,9 +124,9 @@ func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common
|
||||||
receipt.Logs = statedb.GetLogs(tx.Hash())
|
receipt.Logs = statedb.GetLogs(tx.Hash())
|
||||||
receipt.Bloom = types.CreateBloom(types.Receipts{receipt})
|
receipt.Bloom = types.CreateBloom(types.Receipts{receipt})
|
||||||
|
|
||||||
retData := types.ReturnData{receipt.TxHash,data,false}
|
retData := types.ReturnData{TxHash: receipt.TxHash, Data: data, Removed: false}
|
||||||
txPostEvent := TransactionEvent{receipt.TxHash,retData}
|
txPostEvent := TransactionEvent{TxHash: receipt.TxHash, RetData: retData}
|
||||||
go bc.txPostFeed.Send(&txPostEvent)
|
go bc.txPostFeed.Send(&txPostEvent)
|
||||||
|
|
||||||
return receipt, gas, err
|
return receipt, gas, err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -82,14 +82,14 @@ type txdataMarshaling struct {
|
||||||
|
|
||||||
// Return data generated by a transaction which calls a contract method
|
// Return data generated by a transaction which calls a contract method
|
||||||
type ReturnData struct {
|
type ReturnData struct {
|
||||||
// hash of transaction
|
// hash of transaction
|
||||||
TxHash common.Hash `json:"transactionHash" gencodec:"required"`
|
TxHash common.Hash `json:"transactionHash" gencodec:"required"`
|
||||||
|
|
||||||
// address of contract called by transaction
|
// address of contract called by transaction
|
||||||
Data []byte `json:"returndata" gencodec:"required"`
|
Data []byte `json:"returndata" gencodec:"required"`
|
||||||
|
|
||||||
// True if this transaction was reverted due to chain reorg
|
// True if this transaction was reverted due to chain reorg
|
||||||
Removed bool `json:"removed"`
|
Removed bool `json:"removed"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewTransaction(nonce uint64, to common.Address, amount *big.Int, gasLimit uint64, gasPrice *big.Int, data []byte) *Transaction {
|
func NewTransaction(nonce uint64, to common.Address, amount *big.Int, gasLimit uint64, gasPrice *big.Int, data []byte) *Transaction {
|
||||||
|
|
|
||||||
|
|
@ -22,17 +22,17 @@ import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"math/big"
|
"math/big"
|
||||||
|
"reflect"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
"reflect"
|
|
||||||
|
|
||||||
ethereum "github.com/ethereum/go-ethereum"
|
ethereum "github.com/ethereum/go-ethereum"
|
||||||
"github.com/ethereum/go-ethereum/common"
|
"github.com/ethereum/go-ethereum/common"
|
||||||
"github.com/ethereum/go-ethereum/common/hexutil"
|
"github.com/ethereum/go-ethereum/common/hexutil"
|
||||||
"github.com/ethereum/go-ethereum/core/types"
|
"github.com/ethereum/go-ethereum/core/types"
|
||||||
"github.com/ethereum/go-ethereum/log"
|
|
||||||
"github.com/ethereum/go-ethereum/ethdb"
|
"github.com/ethereum/go-ethereum/ethdb"
|
||||||
"github.com/ethereum/go-ethereum/event"
|
"github.com/ethereum/go-ethereum/event"
|
||||||
|
"github.com/ethereum/go-ethereum/log"
|
||||||
"github.com/ethereum/go-ethereum/rpc"
|
"github.com/ethereum/go-ethereum/rpc"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -48,7 +48,7 @@ type filter struct {
|
||||||
hashes []common.Hash
|
hashes []common.Hash
|
||||||
crit FilterCriteria
|
crit FilterCriteria
|
||||||
logs []*types.Log
|
logs []*types.Log
|
||||||
retdata []types.ReturnData
|
retdata []types.ReturnData
|
||||||
s *Subscription // associated subscription in event system
|
s *Subscription // associated subscription in event system
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -233,22 +233,22 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er
|
||||||
}
|
}
|
||||||
|
|
||||||
func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID {
|
func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID {
|
||||||
var (
|
var (
|
||||||
retCh = make(chan *types.ReturnData)
|
retCh = make(chan *types.ReturnData)
|
||||||
retSub = api.events.SubscribeReturnData(retCh)
|
retSub = api.events.SubscribeReturnData(retCh)
|
||||||
)
|
)
|
||||||
|
|
||||||
api.filtersMu.Lock()
|
api.filtersMu.Lock()
|
||||||
api.filters[retSub.ID] = &filter{typ: ReturnDataSubscription, deadline: time.NewTimer(deadline), retdata: make([]types.ReturnData, 0), s: retSub}
|
api.filters[retSub.ID] = &filter{typ: ReturnDataSubscription, deadline: time.NewTimer(deadline), retdata: make([]types.ReturnData, 0), s: retSub}
|
||||||
api.filtersMu.Unlock()
|
api.filtersMu.Unlock()
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case retdata := <-retCh:
|
case retdata := <-retCh:
|
||||||
api.filtersMu.Lock()
|
api.filtersMu.Lock()
|
||||||
if f, found := api.filters[retSub.ID]; found {
|
if f, found := api.filters[retSub.ID]; found {
|
||||||
f.retdata = append(f.retdata, *retdata)
|
f.retdata = append(f.retdata, *retdata)
|
||||||
}
|
}
|
||||||
api.filtersMu.Unlock()
|
api.filtersMu.Unlock()
|
||||||
case <-retSub.Err():
|
case <-retSub.Err():
|
||||||
|
|
@ -257,22 +257,22 @@ func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID {
|
||||||
api.filtersMu.Unlock()
|
api.filtersMu.Unlock()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
return retSub.ID
|
return retSub.ID
|
||||||
}
|
}
|
||||||
|
|
||||||
func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, error) {
|
func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, error) {
|
||||||
notifier, supported := rpc.NotifierFromContext(ctx)
|
notifier, supported := rpc.NotifierFromContext(ctx)
|
||||||
if !supported {
|
if !supported {
|
||||||
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
|
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
|
||||||
}
|
}
|
||||||
|
|
||||||
txListen := make(map[common.Hash]bool)
|
txListen := make(map[common.Hash]bool)
|
||||||
|
|
||||||
rpcSub := notifier.CreateSubscription()
|
rpcSub := notifier.CreateSubscription()
|
||||||
rpcSub.SetType(rpc.ReturnDataSubscription)
|
rpcSub.SetType(rpc.ReturnDataSubscription)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
retCh := make(chan *types.ReturnData)
|
retCh := make(chan *types.ReturnData)
|
||||||
|
|
@ -280,20 +280,20 @@ func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription,
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case msg := <-rpcSub.Update():
|
case msg := <-rpcSub.Update():
|
||||||
// client submitted new tx, save hash for later
|
// client submitted new tx, save hash for later
|
||||||
hash,ishash := msg.(common.Hash)
|
hash, ishash := msg.(common.Hash)
|
||||||
if ishash {
|
if ishash {
|
||||||
txListen[hash] = true
|
txListen[hash] = true
|
||||||
} else {
|
} else {
|
||||||
log.Warn(fmt.Sprintf("Received update msg of invalid type %s for ReturnData subscription",reflect.TypeOf(msg).String()))
|
log.Warn(fmt.Sprintf("Received update msg of invalid type %s for ReturnData subscription", reflect.TypeOf(msg).String()))
|
||||||
}
|
}
|
||||||
case retdata := <-retCh:
|
case retdata := <-retCh:
|
||||||
if txListen[retdata.TxHash] {
|
if txListen[retdata.TxHash] {
|
||||||
// tx from client just completed execution--send back return data
|
// tx from client just completed execution--send back return data
|
||||||
notifier.Notify(rpcSub.ID, retdata)
|
notifier.Notify(rpcSub.ID, retdata)
|
||||||
}
|
}
|
||||||
case <-rpcSub.Err():
|
case <-rpcSub.Err():
|
||||||
retSub.Unsubscribe()
|
retSub.Unsubscribe()
|
||||||
return
|
return
|
||||||
case <-notifier.Closed():
|
case <-notifier.Closed():
|
||||||
|
|
@ -306,7 +306,6 @@ func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription,
|
||||||
return rpcSub, nil
|
return rpcSub, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
// Logs creates a subscription that fires for all new log that match the given filter criteria.
|
// 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) {
|
func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
|
||||||
notifier, supported := rpc.NotifierFromContext(ctx)
|
notifier, supported := rpc.NotifierFromContext(ctx)
|
||||||
|
|
@ -498,10 +497,10 @@ func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
|
||||||
logs := f.logs
|
logs := f.logs
|
||||||
f.logs = nil
|
f.logs = nil
|
||||||
return returnLogs(logs), nil
|
return returnLogs(logs), nil
|
||||||
case ReturnDataSubscription:
|
case ReturnDataSubscription:
|
||||||
retdata := f.retdata
|
retdata := f.retdata
|
||||||
f.retdata = nil
|
f.retdata = nil
|
||||||
return returnRetData(retdata), nil
|
return returnRetData(retdata), nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -527,14 +526,14 @@ func returnLogs(logs []*types.Log) []*types.Log {
|
||||||
}
|
}
|
||||||
|
|
||||||
func returnRetData(rdata []types.ReturnData) []types.ReturnData {
|
func returnRetData(rdata []types.ReturnData) []types.ReturnData {
|
||||||
if rdata == nil {
|
if rdata == nil {
|
||||||
return []types.ReturnData{}
|
return []types.ReturnData{}
|
||||||
} else {
|
} else {
|
||||||
for _,r := range rdata {
|
for _, r := range rdata {
|
||||||
r.Data = []byte{}
|
r.Data = []byte{}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return rdata
|
return rdata
|
||||||
}
|
}
|
||||||
|
|
||||||
// UnmarshalJSON sets *args fields with given data.
|
// UnmarshalJSON sets *args fields with given data.
|
||||||
|
|
|
||||||
|
|
@ -130,7 +130,7 @@ func benchmarkBloomBits(b *testing.B, sectionSize uint64) {
|
||||||
if i%20 == 0 {
|
if i%20 == 0 {
|
||||||
db.Close()
|
db.Close()
|
||||||
db, _ = ethdb.NewLDBDatabase(benchDataDir, 128, 1024)
|
db, _ = ethdb.NewLDBDatabase(benchDataDir, 128, 1024)
|
||||||
backend = &testBackend{mux, db, cnt, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
|
backend = &testBackend{mux, db, cnt, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
|
||||||
}
|
}
|
||||||
var addr common.Address
|
var addr common.Address
|
||||||
addr[0] = byte(i)
|
addr[0] = byte(i)
|
||||||
|
|
@ -191,7 +191,7 @@ func BenchmarkNoBloomBits(b *testing.B) {
|
||||||
fmt.Println("Running filter benchmarks...")
|
fmt.Println("Running filter benchmarks...")
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
mux := new(event.TypeMux)
|
mux := new(event.TypeMux)
|
||||||
backend := &testBackend{mux, db, 0, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
|
backend := &testBackend{mux, db, 0, new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed), new(event.Feed)}
|
||||||
filter := New(backend, 0, int64(headNum), []common.Address{{}}, nil)
|
filter := New(backend, 0, int64(headNum), []common.Address{{}}, nil)
|
||||||
filter.Logs(context.Background())
|
filter.Logs(context.Background())
|
||||||
d := time.Since(start)
|
d := time.Since(start)
|
||||||
|
|
|
||||||
|
|
@ -37,7 +37,7 @@ type Backend interface {
|
||||||
GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
|
GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
|
||||||
|
|
||||||
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
|
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
|
||||||
SubscribeTransactionEvent(chan<- *core.TransactionEvent) event.Subscription
|
SubscribeTransactionEvent(chan<- *core.TransactionEvent) event.Subscription
|
||||||
SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
|
SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
|
||||||
SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
|
SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
|
||||||
SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
|
SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription
|
||||||
|
|
|
||||||
|
|
@ -51,9 +51,9 @@ const (
|
||||||
PendingTransactionsSubscription Type = rpc.PendingTransactionsSubscription
|
PendingTransactionsSubscription Type = rpc.PendingTransactionsSubscription
|
||||||
// BlocksSubscription queries hashes for blocks that are imported
|
// BlocksSubscription queries hashes for blocks that are imported
|
||||||
BlocksSubscription Type = rpc.BlocksSubscription
|
BlocksSubscription Type = rpc.BlocksSubscription
|
||||||
// ReturnData queries for return data from transactions executed by a
|
// ReturnData queries for return data from transactions executed by a
|
||||||
// particular rpc client
|
// particular rpc client
|
||||||
ReturnDataSubscription Type = rpc.ReturnDataSubscription
|
ReturnDataSubscription Type = rpc.ReturnDataSubscription
|
||||||
// LastSubscription keeps track of the last index
|
// LastSubscription keeps track of the last index
|
||||||
LastIndexSubscription Type = rpc.LastIndexSubscription
|
LastIndexSubscription Type = rpc.LastIndexSubscription
|
||||||
)
|
)
|
||||||
|
|
@ -64,7 +64,7 @@ const (
|
||||||
// The number is referenced from the size of tx pool.
|
// The number is referenced from the size of tx pool.
|
||||||
txPreChanSize = 4096
|
txPreChanSize = 4096
|
||||||
// txPostChanSize is the size of channel listening to TransactionEvent.
|
// txPostChanSize is the size of channel listening to TransactionEvent.
|
||||||
// For now, setting to same as txPreChanSize--good number?
|
// For now, setting to same as txPreChanSize--good number?
|
||||||
txPostChanSize = 4096
|
txPostChanSize = 4096
|
||||||
// rmLogsChanSize is the size of channel listening to RemovedLogsEvent.
|
// rmLogsChanSize is the size of channel listening to RemovedLogsEvent.
|
||||||
rmLogsChanSize = 10
|
rmLogsChanSize = 10
|
||||||
|
|
@ -79,16 +79,16 @@ var (
|
||||||
)
|
)
|
||||||
|
|
||||||
type subscription struct {
|
type subscription struct {
|
||||||
id rpc.ID
|
id rpc.ID
|
||||||
typ Type
|
typ Type
|
||||||
created time.Time
|
created time.Time
|
||||||
logsCrit ethereum.FilterQuery
|
logsCrit ethereum.FilterQuery
|
||||||
logs chan []*types.Log
|
logs chan []*types.Log
|
||||||
hashes chan common.Hash
|
hashes chan common.Hash
|
||||||
headers chan *types.Header
|
headers chan *types.Header
|
||||||
retdata chan *types.ReturnData
|
retdata chan *types.ReturnData
|
||||||
installed chan struct{} // closed when the filter is installed
|
installed chan struct{} // closed when the filter is installed
|
||||||
err chan error // closed when the filter is uninstalled
|
err chan error // closed when the filter is uninstalled
|
||||||
}
|
}
|
||||||
|
|
||||||
// EventSystem creates subscriptions, processes events and broadcasts them to the
|
// EventSystem creates subscriptions, processes events and broadcasts them to the
|
||||||
|
|
@ -292,15 +292,15 @@ func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscr
|
||||||
// SubscribeReturnData creates a subscription that captures return data for transactions
|
// SubscribeReturnData creates a subscription that captures return data for transactions
|
||||||
// executed by a particular rpc client
|
// executed by a particular rpc client
|
||||||
func (es *EventSystem) SubscribeReturnData(retCh chan *types.ReturnData) *Subscription {
|
func (es *EventSystem) SubscribeReturnData(retCh chan *types.ReturnData) *Subscription {
|
||||||
sub := &subscription{
|
sub := &subscription{
|
||||||
id: rpc.NewID(),
|
id: rpc.NewID(),
|
||||||
typ: ReturnDataSubscription,
|
typ: ReturnDataSubscription,
|
||||||
created: time.Now(),
|
created: time.Now(),
|
||||||
retdata: retCh,
|
retdata: retCh,
|
||||||
installed: make(chan struct{}),
|
installed: make(chan struct{}),
|
||||||
err: make(chan error),
|
err: make(chan error),
|
||||||
}
|
}
|
||||||
return es.subscribe(sub)
|
return es.subscribe(sub)
|
||||||
}
|
}
|
||||||
|
|
||||||
type filterIndex map[Type]map[rpc.ID]*subscription
|
type filterIndex map[Type]map[rpc.ID]*subscription
|
||||||
|
|
@ -341,10 +341,10 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
|
||||||
for _, f := range filters[PendingTransactionsSubscription] {
|
for _, f := range filters[PendingTransactionsSubscription] {
|
||||||
f.hashes <- e.Tx.Hash()
|
f.hashes <- e.Tx.Hash()
|
||||||
}
|
}
|
||||||
case *core.TransactionEvent:
|
case *core.TransactionEvent:
|
||||||
for _, f := range filters[ReturnDataSubscription] {
|
for _, f := range filters[ReturnDataSubscription] {
|
||||||
f.retdata <- &e.RetData
|
f.retdata <- &e.RetData
|
||||||
}
|
}
|
||||||
case core.ChainEvent:
|
case core.ChainEvent:
|
||||||
for _, f := range filters[BlocksSubscription] {
|
for _, f := range filters[BlocksSubscription] {
|
||||||
f.headers <- e.Block.Header()
|
f.headers <- e.Block.Header()
|
||||||
|
|
@ -442,9 +442,9 @@ func (es *EventSystem) eventLoop() {
|
||||||
// Subscribe to TxPreEvent from txpool
|
// Subscribe to TxPreEvent from txpool
|
||||||
txPreCh = make(chan core.TxPreEvent, txPreChanSize)
|
txPreCh = make(chan core.TxPreEvent, txPreChanSize)
|
||||||
txPreSub = es.backend.SubscribeTxPreEvent(txPreCh)
|
txPreSub = es.backend.SubscribeTxPreEvent(txPreCh)
|
||||||
// Subscribe to TransactionEvent from applyTransaction
|
// Subscribe to TransactionEvent from applyTransaction
|
||||||
txCh = make(chan *core.TransactionEvent, txPostChanSize)
|
txCh = make(chan *core.TransactionEvent, txPostChanSize)
|
||||||
txSub = es.backend.SubscribeTransactionEvent(txCh)
|
txSub = es.backend.SubscribeTransactionEvent(txCh)
|
||||||
// Subscribe RemovedLogsEvent
|
// Subscribe RemovedLogsEvent
|
||||||
rmLogsCh = make(chan core.RemovedLogsEvent, rmLogsChanSize)
|
rmLogsCh = make(chan core.RemovedLogsEvent, rmLogsChanSize)
|
||||||
rmLogsSub = es.backend.SubscribeRemovedLogsEvent(rmLogsCh)
|
rmLogsSub = es.backend.SubscribeRemovedLogsEvent(rmLogsCh)
|
||||||
|
|
|
||||||
|
|
@ -42,6 +42,7 @@ type testBackend struct {
|
||||||
db ethdb.Database
|
db ethdb.Database
|
||||||
sections uint64
|
sections uint64
|
||||||
txFeed *event.Feed
|
txFeed *event.Feed
|
||||||
|
txPostFeed *event.Feed
|
||||||
rmLogsFeed *event.Feed
|
rmLogsFeed *event.Feed
|
||||||
logsFeed *event.Feed
|
logsFeed *event.Feed
|
||||||
chainFeed *event.Feed
|
chainFeed *event.Feed
|
||||||
|
|
@ -100,6 +101,10 @@ func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subsc
|
||||||
return b.chainFeed.Subscribe(ch)
|
return b.chainFeed.Subscribe(ch)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (b *testBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription {
|
||||||
|
return b.txPostFeed.Subscribe(ch)
|
||||||
|
}
|
||||||
|
|
||||||
func (b *testBackend) BloomStatus() (uint64, uint64) {
|
func (b *testBackend) BloomStatus() (uint64, uint64) {
|
||||||
return params.BloomBitsBlocks, b.sections
|
return params.BloomBitsBlocks, b.sections
|
||||||
}
|
}
|
||||||
|
|
@ -143,10 +148,11 @@ func TestBlockSubscription(t *testing.T) {
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
db, _ = ethdb.NewMemDatabase()
|
db, _ = ethdb.NewMemDatabase()
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
api = NewPublicFilterAPI(backend, false)
|
api = NewPublicFilterAPI(backend, false)
|
||||||
genesis = new(core.Genesis).MustCommit(db)
|
genesis = new(core.Genesis).MustCommit(db)
|
||||||
chain, _ = core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), db, 10, func(i int, gen *core.BlockGen) {})
|
chain, _ = core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), db, 10, func(i int, gen *core.BlockGen) {})
|
||||||
|
|
@ -200,10 +206,11 @@ func TestPendingTxFilter(t *testing.T) {
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
db, _ = ethdb.NewMemDatabase()
|
db, _ = ethdb.NewMemDatabase()
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
api = NewPublicFilterAPI(backend, false)
|
api = NewPublicFilterAPI(backend, false)
|
||||||
|
|
||||||
transactions = []*types.Transaction{
|
transactions = []*types.Transaction{
|
||||||
|
|
@ -263,10 +270,11 @@ func TestLogFilterCreation(t *testing.T) {
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
db, _ = ethdb.NewMemDatabase()
|
db, _ = ethdb.NewMemDatabase()
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
api = NewPublicFilterAPI(backend, false)
|
api = NewPublicFilterAPI(backend, false)
|
||||||
|
|
||||||
testCases = []struct {
|
testCases = []struct {
|
||||||
|
|
@ -312,10 +320,11 @@ func TestInvalidLogFilterCreation(t *testing.T) {
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
db, _ = ethdb.NewMemDatabase()
|
db, _ = ethdb.NewMemDatabase()
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
api = NewPublicFilterAPI(backend, false)
|
api = NewPublicFilterAPI(backend, false)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -342,10 +351,11 @@ func TestLogFilter(t *testing.T) {
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
db, _ = ethdb.NewMemDatabase()
|
db, _ = ethdb.NewMemDatabase()
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
api = NewPublicFilterAPI(backend, false)
|
api = NewPublicFilterAPI(backend, false)
|
||||||
|
|
||||||
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
|
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
|
||||||
|
|
@ -461,10 +471,11 @@ func TestPendingLogsSubscription(t *testing.T) {
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
db, _ = ethdb.NewMemDatabase()
|
db, _ = ethdb.NewMemDatabase()
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
api = NewPublicFilterAPI(backend, false)
|
api = NewPublicFilterAPI(backend, false)
|
||||||
|
|
||||||
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
|
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
|
||||||
|
|
|
||||||
|
|
@ -53,10 +53,11 @@ func BenchmarkFilters(b *testing.B) {
|
||||||
db, _ = ethdb.NewLDBDatabase(dir, 0, 0)
|
db, _ = ethdb.NewLDBDatabase(dir, 0, 0)
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
|
key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
|
||||||
addr1 = crypto.PubkeyToAddress(key1.PublicKey)
|
addr1 = crypto.PubkeyToAddress(key1.PublicKey)
|
||||||
addr2 = common.BytesToAddress([]byte("jeff"))
|
addr2 = common.BytesToAddress([]byte("jeff"))
|
||||||
|
|
@ -118,10 +119,11 @@ func TestFilters(t *testing.T) {
|
||||||
db, _ = ethdb.NewLDBDatabase(dir, 0, 0)
|
db, _ = ethdb.NewLDBDatabase(dir, 0, 0)
|
||||||
mux = new(event.TypeMux)
|
mux = new(event.TypeMux)
|
||||||
txFeed = new(event.Feed)
|
txFeed = new(event.Feed)
|
||||||
|
txPostFeed = new(event.Feed)
|
||||||
rmLogsFeed = new(event.Feed)
|
rmLogsFeed = new(event.Feed)
|
||||||
logsFeed = new(event.Feed)
|
logsFeed = new(event.Feed)
|
||||||
chainFeed = new(event.Feed)
|
chainFeed = new(event.Feed)
|
||||||
backend = &testBackend{mux, db, 0, txFeed, rmLogsFeed, logsFeed, chainFeed}
|
backend = &testBackend{mux, db, 0, txFeed, txPostFeed, rmLogsFeed, logsFeed, chainFeed}
|
||||||
key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
|
key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
|
||||||
addr = crypto.PubkeyToAddress(key1.PublicKey)
|
addr = crypto.PubkeyToAddress(key1.PublicKey)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1185,15 +1185,15 @@ func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (c
|
||||||
log.Info("Submitted transaction", "fullhash", tx.Hash().Hex(), "recipient", tx.To())
|
log.Info("Submitted transaction", "fullhash", tx.Hash().Hex(), "recipient", tx.To())
|
||||||
}
|
}
|
||||||
|
|
||||||
notifier,supported := rpc.NotifierFromContext(ctx)
|
notifier, supported := rpc.NotifierFromContext(ctx)
|
||||||
if supported {
|
if supported {
|
||||||
// If this client has a returnData subscription, add tx hash to set of transactions
|
// If this client has a returnData subscription, add tx hash to set of transactions
|
||||||
// whose return data should be sent back to rpc subscriber after transaction
|
// whose return data should be sent back to rpc subscriber after transaction
|
||||||
// is sealed in a new block.
|
// is sealed in a new block.
|
||||||
go notifier.UpdateSubscriptions(rpc.ReturnDataSubscription,tx.Hash())
|
go notifier.UpdateSubscriptions(rpc.ReturnDataSubscription, tx.Hash())
|
||||||
}
|
}
|
||||||
|
|
||||||
return tx.Hash(), nil
|
return tx.Hash(), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SendTransaction creates a transaction for the given argument, sign it and submit it to the
|
// SendTransaction creates a transaction for the given argument, sign it and submit it to the
|
||||||
|
|
|
||||||
|
|
@ -57,7 +57,7 @@ type Backend interface {
|
||||||
SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
|
SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
|
||||||
SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
|
SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
|
||||||
SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription
|
SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription
|
||||||
SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription
|
SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription
|
||||||
|
|
||||||
// TxPool API
|
// TxPool API
|
||||||
SendTx(ctx context.Context, signedTx *types.Transaction) error
|
SendTx(ctx context.Context, signedTx *types.Transaction) error
|
||||||
|
|
|
||||||
|
|
@ -52,7 +52,7 @@ type LightChain struct {
|
||||||
chainFeed event.Feed
|
chainFeed event.Feed
|
||||||
chainSideFeed event.Feed
|
chainSideFeed event.Feed
|
||||||
chainHeadFeed event.Feed
|
chainHeadFeed event.Feed
|
||||||
txPostFeed event.Feed
|
txPostFeed event.Feed
|
||||||
scope event.SubscriptionScope
|
scope event.SubscriptionScope
|
||||||
genesisBlock *types.Block
|
genesisBlock *types.Block
|
||||||
|
|
||||||
|
|
@ -494,7 +494,6 @@ func (self *LightChain) SubscribeTransactionEvent(ch chan<- *core.TransactionEve
|
||||||
return self.scope.Track(self.txPostFeed.Subscribe(ch))
|
return self.scope.Track(self.txPostFeed.Subscribe(ch))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
// SubscribeLogsEvent implements the interface of filters.Backend
|
// SubscribeLogsEvent implements the interface of filters.Backend
|
||||||
// LightChain does not send logs events, so return an empty subscription.
|
// LightChain does not send logs events, so return an empty subscription.
|
||||||
func (self *LightChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
|
func (self *LightChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
|
||||||
|
|
|
||||||
|
|
@ -26,15 +26,15 @@ var (
|
||||||
// ErrNotificationsUnsupported is returned when the connection doesn't support notifications
|
// ErrNotificationsUnsupported is returned when the connection doesn't support notifications
|
||||||
ErrNotificationsUnsupported = errors.New("notifications not supported")
|
ErrNotificationsUnsupported = errors.New("notifications not supported")
|
||||||
// ErrNotificationNotFound is returned when the notification for the given id is not found
|
// ErrNotificationNotFound is returned when the notification for the given id is not found
|
||||||
ErrSubscriptionNotFound = errors.New("subscription not found")
|
ErrSubscriptionNotFound = errors.New("subscription not found")
|
||||||
ErrNamespaceNotFound = errors.New("namespace not found")
|
ErrNamespaceNotFound = errors.New("namespace not found")
|
||||||
ErrSubscriptionTypeNotFound = errors.New("no active subscriptions of specified type")
|
ErrSubscriptionTypeNotFound = errors.New("no active subscriptions of specified type")
|
||||||
)
|
)
|
||||||
|
|
||||||
// ID defines a pseudo random number that is used to identify RPC subscriptions.
|
// ID defines a pseudo random number that is used to identify RPC subscriptions.
|
||||||
type ID string
|
type ID string
|
||||||
|
|
||||||
// SubscriptionType determines the type of subscription (which allows sending updates to
|
// SubscriptionType determines the type of subscription (which allows sending updates to
|
||||||
// subscriptions of a particular type); this is also used by eth/filters to index different
|
// subscriptions of a particular type); this is also used by eth/filters to index different
|
||||||
// filter types.
|
// filter types.
|
||||||
type SubscriptionType byte
|
type SubscriptionType byte
|
||||||
|
|
@ -53,31 +53,29 @@ const (
|
||||||
PendingTransactionsSubscription
|
PendingTransactionsSubscription
|
||||||
// BlocksSubscription queries hashes for blocks that are imported
|
// BlocksSubscription queries hashes for blocks that are imported
|
||||||
BlocksSubscription
|
BlocksSubscription
|
||||||
// ReturnData queries for return data from transactions executed by a
|
// ReturnData queries for return data from transactions executed by a
|
||||||
// particular rpc client
|
// particular rpc client
|
||||||
ReturnDataSubscription
|
ReturnDataSubscription
|
||||||
// LastSubscription keeps track of the last index
|
// LastSubscription keeps track of the last index
|
||||||
LastIndexSubscription
|
LastIndexSubscription
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// a Subscription is created by a notifier and tied to that notifier. The client can use
|
// a Subscription is created by a notifier and tied to that notifier. The client can use
|
||||||
// this subscription to wait for an unsubscribe request for the client, see Err().
|
// this subscription to wait for an unsubscribe request for the client, see Err().
|
||||||
type Subscription struct {
|
type Subscription struct {
|
||||||
ID ID
|
ID ID
|
||||||
namespace string
|
namespace string
|
||||||
Type SubscriptionType
|
Type SubscriptionType
|
||||||
update chan interface{} // used to send update filter criteria of subscription
|
update chan interface{} // used to send update filter criteria of subscription
|
||||||
err chan error // closed on unsubscribe
|
err chan error // closed on unsubscribe
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Subscription) SetType(subType SubscriptionType) {
|
func (s *Subscription) SetType(subType SubscriptionType) {
|
||||||
s.Type = subType
|
s.Type = subType
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Subscription) Update() <-chan interface{} {
|
func (s *Subscription) Update() <-chan interface{} {
|
||||||
return s.update
|
return s.update
|
||||||
}
|
}
|
||||||
|
|
||||||
// Err returns a channel that is closed when the client sends an unsubscribe request.
|
// Err returns a channel that is closed when the client sends an unsubscribe request.
|
||||||
|
|
@ -118,7 +116,7 @@ func NotifierFromContext(ctx context.Context) (*Notifier, bool) {
|
||||||
// are dropped until the subscription is marked as active. This is done
|
// are dropped until the subscription is marked as active. This is done
|
||||||
// by the RPC server after the subscription ID is send to the client.
|
// by the RPC server after the subscription ID is send to the client.
|
||||||
func (n *Notifier) CreateSubscription() *Subscription {
|
func (n *Notifier) CreateSubscription() *Subscription {
|
||||||
s := &Subscription{ID: NewID(), update: make(chan interface{}),err: make(chan error)}
|
s := &Subscription{ID: NewID(), update: make(chan interface{}), err: make(chan error)}
|
||||||
n.subMu.Lock()
|
n.subMu.Lock()
|
||||||
n.inactive[s.ID] = s
|
n.inactive[s.ID] = s
|
||||||
n.subMu.Unlock()
|
n.subMu.Unlock()
|
||||||
|
|
@ -148,15 +146,15 @@ func (n *Notifier) Notify(id ID, data interface{}) error {
|
||||||
// generate notifcations.
|
// generate notifcations.
|
||||||
//
|
//
|
||||||
func (n *Notifier) UpdateSubscription(id ID, message interface{}) error {
|
func (n *Notifier) UpdateSubscription(id ID, message interface{}) error {
|
||||||
n.subMu.RLock()
|
n.subMu.RLock()
|
||||||
defer n.subMu.RUnlock()
|
defer n.subMu.RUnlock()
|
||||||
sub, validid := n.active[id]
|
sub, validid := n.active[id]
|
||||||
if validid {
|
if validid {
|
||||||
sub.update <- message
|
sub.update <- message
|
||||||
return nil
|
return nil
|
||||||
} else {
|
} else {
|
||||||
return ErrSubscriptionNotFound
|
return ErrSubscriptionNotFound
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Send an update message to the update channels of all subscriptions of
|
// Send an update message to the update channels of all subscriptions of
|
||||||
|
|
@ -170,16 +168,16 @@ func (n *Notifier) UpdateSubscription(id ID, message interface{}) error {
|
||||||
// submits a new transaction, this is used to add the tx to the list of transactions
|
// submits a new transaction, this is used to add the tx to the list of transactions
|
||||||
// which should generate a TransactionEvent and notify the client with return data.
|
// which should generate a TransactionEvent and notify the client with return data.
|
||||||
func (n *Notifier) UpdateSubscriptions(subType SubscriptionType, message interface{}) error {
|
func (n *Notifier) UpdateSubscriptions(subType SubscriptionType, message interface{}) error {
|
||||||
n.subMu.RLock()
|
n.subMu.RLock()
|
||||||
defer n.subMu.RUnlock()
|
defer n.subMu.RUnlock()
|
||||||
for _,sub := range n.active {
|
for _, sub := range n.active {
|
||||||
if sub.Type == subType {
|
if sub.Type == subType {
|
||||||
sub.update <- message
|
sub.update <- message
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return ErrSubscriptionTypeNotFound
|
return ErrSubscriptionTypeNotFound
|
||||||
}
|
}
|
||||||
|
|
||||||
// Closed returns a channel that is closed when the RPC connection is closed.
|
// Closed returns a channel that is closed when the RPC connection is closed.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue