From 169c590b07711f08512184fc19faa1f94d263802 Mon Sep 17 00:00:00 2001 From: Domino Valdano Date: Sat, 31 Mar 2018 15:23:25 -0700 Subject: [PATCH] eth/filters,rpc,core,internal/ethapi: Fixed formating, linting, vet issues. --- accounts/abi/bind/backends/simulated.go | 4 ++ core/blockchain.go | 6 +- core/events.go | 6 +- core/state_processor.go | 6 +- core/types/transaction.go | 14 ++-- eth/filters/api.go | 85 ++++++++++++------------- eth/filters/bench_test.go | 4 +- eth/filters/filter.go | 2 +- eth/filters/filter_system.go | 50 +++++++-------- eth/filters/filter_system_test.go | 23 +++++-- eth/filters/filter_test.go | 6 +- internal/ethapi/api.go | 16 ++--- internal/ethapi/backend.go | 2 +- light/lightchain.go | 3 +- rpc/subscription.go | 62 +++++++++--------- 15 files changed, 151 insertions(+), 138 deletions(-) diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index 2b5c5fc4a4..a28bd6dcdc 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -461,6 +461,10 @@ func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscr 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) ServiceFilter(ctx context.Context, ms *bloombits.MatcherSession) { panic("not supported") diff --git a/core/blockchain.go b/core/blockchain.go index fc60463d0e..44e3adf0ff 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -99,7 +99,7 @@ type BlockChain struct { chainSideFeed event.Feed chainHeadFeed 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 genesisBlock *types.Block @@ -1372,8 +1372,8 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) { case ChainSideEvent: bc.chainSideFeed.Send(ev) - } - } + } + } } func (bc *BlockChain) update() { diff --git a/core/events.go b/core/events.go index 12723f0df8..7c1a683920 100644 --- a/core/events.go +++ b/core/events.go @@ -36,9 +36,9 @@ type PendingStateEvent struct{} type NewMinedBlockEvent struct{ Block *types.Block } // TransactionEvent is posted when a transaction completes execution -type TransactionEvent struct{ - TxHash common.Hash - RetData types.ReturnData +type TransactionEvent struct { + TxHash common.Hash + RetData types.ReturnData } // RemovedTransactionEvent is posted when a reorg happens diff --git a/core/state_processor.go b/core/state_processor.go index cf0fc494f0..fa5edbf550 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -124,9 +124,9 @@ func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common receipt.Logs = statedb.GetLogs(tx.Hash()) receipt.Bloom = types.CreateBloom(types.Receipts{receipt}) - retData := types.ReturnData{receipt.TxHash,data,false} - txPostEvent := TransactionEvent{receipt.TxHash,retData} - go bc.txPostFeed.Send(&txPostEvent) + retData := types.ReturnData{TxHash: receipt.TxHash, Data: data, Removed: false} + txPostEvent := TransactionEvent{TxHash: receipt.TxHash, RetData: retData} + go bc.txPostFeed.Send(&txPostEvent) return receipt, gas, err } diff --git a/core/types/transaction.go b/core/types/transaction.go index 7b387aa996..9fd8899319 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -82,14 +82,14 @@ type txdataMarshaling struct { // Return data generated by a transaction which calls a contract method type ReturnData struct { - // hash of transaction - TxHash common.Hash `json:"transactionHash" gencodec:"required"` + // hash of transaction + TxHash common.Hash `json:"transactionHash" gencodec:"required"` - // address of contract called by transaction - Data []byte `json:"returndata" gencodec:"required"` - - // True if this transaction was reverted due to chain reorg - Removed bool `json:"removed"` + // address of contract called by transaction + Data []byte `json:"returndata" gencodec:"required"` + + // True if this transaction was reverted due to chain reorg + Removed bool `json:"removed"` } func NewTransaction(nonce uint64, to common.Address, amount *big.Int, gasLimit uint64, gasPrice *big.Int, data []byte) *Transaction { diff --git a/eth/filters/api.go b/eth/filters/api.go index fbe2a88078..950d1bdcd7 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -22,17 +22,17 @@ import ( "errors" "fmt" "math/big" + "reflect" "sync" "time" - "reflect" ethereum "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rpc" ) @@ -48,7 +48,7 @@ type filter struct { hashes []common.Hash crit FilterCriteria logs []*types.Log - retdata []types.ReturnData + retdata []types.ReturnData 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 { - var ( - retCh = make(chan *types.ReturnData) - retSub = api.events.SubscribeReturnData(retCh) - ) + var ( + retCh = make(chan *types.ReturnData) + 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.filtersMu.Unlock() - go func() { - for { + go func() { + for { select { case retdata := <-retCh: api.filtersMu.Lock() if f, found := api.filters[retSub.ID]; found { - f.retdata = append(f.retdata, *retdata) + f.retdata = append(f.retdata, *retdata) } api.filtersMu.Unlock() case <-retSub.Err(): @@ -257,22 +257,22 @@ func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID { api.filtersMu.Unlock() return } - } + } }() - return retSub.ID + return retSub.ID } func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) + notifier, supported := rpc.NotifierFromContext(ctx) if !supported { return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported } - txListen := make(map[common.Hash]bool) + txListen := make(map[common.Hash]bool) rpcSub := notifier.CreateSubscription() - rpcSub.SetType(rpc.ReturnDataSubscription) + rpcSub.SetType(rpc.ReturnDataSubscription) go func() { retCh := make(chan *types.ReturnData) @@ -280,20 +280,20 @@ func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, for { select { - case msg := <-rpcSub.Update(): - // client submitted new tx, save hash for later - hash,ishash := msg.(common.Hash) - if ishash { - txListen[hash] = true - } else { - log.Warn(fmt.Sprintf("Received update msg of invalid type %s for ReturnData subscription",reflect.TypeOf(msg).String())) - } - case retdata := <-retCh: - if txListen[retdata.TxHash] { - // tx from client just completed execution--send back return data - notifier.Notify(rpcSub.ID, retdata) - } - case <-rpcSub.Err(): + case msg := <-rpcSub.Update(): + // client submitted new tx, save hash for later + hash, ishash := msg.(common.Hash) + if ishash { + txListen[hash] = true + } else { + log.Warn(fmt.Sprintf("Received update msg of invalid type %s for ReturnData subscription", reflect.TypeOf(msg).String())) + } + case retdata := <-retCh: + if txListen[retdata.TxHash] { + // tx from client just completed execution--send back return data + notifier.Notify(rpcSub.ID, retdata) + } + case <-rpcSub.Err(): retSub.Unsubscribe() return case <-notifier.Closed(): @@ -306,7 +306,6 @@ func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, return rpcSub, nil } - // Logs creates a subscription that fires for all new log that match the given filter criteria. func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) @@ -498,10 +497,10 @@ func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) { logs := f.logs f.logs = nil return returnLogs(logs), nil - case ReturnDataSubscription: - retdata := f.retdata - f.retdata = nil - return returnRetData(retdata), nil + case ReturnDataSubscription: + retdata := f.retdata + f.retdata = nil + return returnRetData(retdata), nil } } @@ -527,14 +526,14 @@ func returnLogs(logs []*types.Log) []*types.Log { } func returnRetData(rdata []types.ReturnData) []types.ReturnData { - if rdata == nil { - return []types.ReturnData{} - } else { - for _,r := range rdata { - r.Data = []byte{} - } - } - return rdata + if rdata == nil { + return []types.ReturnData{} + } else { + for _, r := range rdata { + r.Data = []byte{} + } + } + return rdata } // UnmarshalJSON sets *args fields with given data. diff --git a/eth/filters/bench_test.go b/eth/filters/bench_test.go index 0a0929bc10..d2aaf42a9e 100644 --- a/eth/filters/bench_test.go +++ b/eth/filters/bench_test.go @@ -130,7 +130,7 @@ func benchmarkBloomBits(b *testing.B, sectionSize uint64) { if i%20 == 0 { db.Close() 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 addr[0] = byte(i) @@ -191,7 +191,7 @@ func BenchmarkNoBloomBits(b *testing.B) { fmt.Println("Running filter benchmarks...") start := time.Now() 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.Logs(context.Background()) d := time.Since(start) diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 964ff366a0..085a38aeb5 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -37,7 +37,7 @@ type Backend interface { GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error) 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 SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index 78a648a6ac..daddfd6fcf 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -51,9 +51,9 @@ const ( PendingTransactionsSubscription Type = rpc.PendingTransactionsSubscription // BlocksSubscription queries hashes for blocks that are imported BlocksSubscription Type = rpc.BlocksSubscription - // ReturnData queries for return data from transactions executed by a - // particular rpc client - ReturnDataSubscription Type = rpc.ReturnDataSubscription + // ReturnData queries for return data from transactions executed by a + // particular rpc client + ReturnDataSubscription Type = rpc.ReturnDataSubscription // LastSubscription keeps track of the last index LastIndexSubscription Type = rpc.LastIndexSubscription ) @@ -64,7 +64,7 @@ const ( // The number is referenced from the size of tx pool. txPreChanSize = 4096 // 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 // rmLogsChanSize is the size of channel listening to RemovedLogsEvent. rmLogsChanSize = 10 @@ -79,16 +79,16 @@ var ( ) type subscription struct { - id rpc.ID - typ Type - created time.Time - logsCrit ethereum.FilterQuery - logs chan []*types.Log - hashes chan common.Hash - headers chan *types.Header - retdata chan *types.ReturnData - installed chan struct{} // closed when the filter is installed - err chan error // closed when the filter is uninstalled + id rpc.ID + typ Type + created time.Time + logsCrit ethereum.FilterQuery + logs chan []*types.Log + hashes chan common.Hash + headers chan *types.Header + retdata chan *types.ReturnData + installed chan struct{} // closed when the filter is installed + err chan error // closed when the filter is uninstalled } // 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 // executed by a particular rpc client func (es *EventSystem) SubscribeReturnData(retCh chan *types.ReturnData) *Subscription { - sub := &subscription{ + sub := &subscription{ id: rpc.NewID(), typ: ReturnDataSubscription, created: time.Now(), - retdata: retCh, + retdata: retCh, installed: make(chan struct{}), err: make(chan error), - } - return es.subscribe(sub) + } + return es.subscribe(sub) } 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] { f.hashes <- e.Tx.Hash() } - case *core.TransactionEvent: - for _, f := range filters[ReturnDataSubscription] { - f.retdata <- &e.RetData - } + case *core.TransactionEvent: + for _, f := range filters[ReturnDataSubscription] { + f.retdata <- &e.RetData + } case core.ChainEvent: for _, f := range filters[BlocksSubscription] { f.headers <- e.Block.Header() @@ -442,9 +442,9 @@ func (es *EventSystem) eventLoop() { // Subscribe to TxPreEvent from txpool txPreCh = make(chan core.TxPreEvent, txPreChanSize) txPreSub = es.backend.SubscribeTxPreEvent(txPreCh) - // Subscribe to TransactionEvent from applyTransaction - txCh = make(chan *core.TransactionEvent, txPostChanSize) - txSub = es.backend.SubscribeTransactionEvent(txCh) + // Subscribe to TransactionEvent from applyTransaction + txCh = make(chan *core.TransactionEvent, txPostChanSize) + txSub = es.backend.SubscribeTransactionEvent(txCh) // Subscribe RemovedLogsEvent rmLogsCh = make(chan core.RemovedLogsEvent, rmLogsChanSize) rmLogsSub = es.backend.SubscribeRemovedLogsEvent(rmLogsCh) diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index 61761151a2..6f80e28c4b 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -42,6 +42,7 @@ type testBackend struct { db ethdb.Database sections uint64 txFeed *event.Feed + txPostFeed *event.Feed rmLogsFeed *event.Feed logsFeed *event.Feed chainFeed *event.Feed @@ -100,6 +101,10 @@ func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subsc 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) { return params.BloomBitsBlocks, b.sections } @@ -143,10 +148,11 @@ func TestBlockSubscription(t *testing.T) { mux = new(event.TypeMux) db, _ = ethdb.NewMemDatabase() txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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) genesis = new(core.Genesis).MustCommit(db) 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) db, _ = ethdb.NewMemDatabase() txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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) transactions = []*types.Transaction{ @@ -263,10 +270,11 @@ func TestLogFilterCreation(t *testing.T) { mux = new(event.TypeMux) db, _ = ethdb.NewMemDatabase() txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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) testCases = []struct { @@ -312,10 +320,11 @@ func TestInvalidLogFilterCreation(t *testing.T) { mux = new(event.TypeMux) db, _ = ethdb.NewMemDatabase() txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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) ) @@ -342,10 +351,11 @@ func TestLogFilter(t *testing.T) { mux = new(event.TypeMux) db, _ = ethdb.NewMemDatabase() txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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) firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") @@ -461,10 +471,11 @@ func TestPendingLogsSubscription(t *testing.T) { mux = new(event.TypeMux) db, _ = ethdb.NewMemDatabase() txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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) firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go index 0018142c46..8c531f93b5 100644 --- a/eth/filters/filter_test.go +++ b/eth/filters/filter_test.go @@ -53,10 +53,11 @@ func BenchmarkFilters(b *testing.B) { db, _ = ethdb.NewLDBDatabase(dir, 0, 0) mux = new(event.TypeMux) txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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") addr1 = crypto.PubkeyToAddress(key1.PublicKey) addr2 = common.BytesToAddress([]byte("jeff")) @@ -118,10 +119,11 @@ func TestFilters(t *testing.T) { db, _ = ethdb.NewLDBDatabase(dir, 0, 0) mux = new(event.TypeMux) txFeed = new(event.Feed) + txPostFeed = new(event.Feed) rmLogsFeed = new(event.Feed) logsFeed = 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") addr = crypto.PubkeyToAddress(key1.PublicKey) diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index e799b3f95b..2463682612 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -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()) } - notifier,supported := rpc.NotifierFromContext(ctx) - if supported { - // 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 - // is sealed in a new block. - go notifier.UpdateSubscriptions(rpc.ReturnDataSubscription,tx.Hash()) - } + notifier, supported := rpc.NotifierFromContext(ctx) + if supported { + // 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 + // is sealed in a new block. + 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 diff --git a/internal/ethapi/backend.go b/internal/ethapi/backend.go index b9645f57b9..40f05d2890 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -57,7 +57,7 @@ type Backend interface { SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) 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 SendTx(ctx context.Context, signedTx *types.Transaction) error diff --git a/light/lightchain.go b/light/lightchain.go index eaa14195ca..640ee3b954 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -52,7 +52,7 @@ type LightChain struct { chainFeed event.Feed chainSideFeed event.Feed chainHeadFeed event.Feed - txPostFeed event.Feed + txPostFeed event.Feed scope event.SubscriptionScope genesisBlock *types.Block @@ -494,7 +494,6 @@ func (self *LightChain) SubscribeTransactionEvent(ch chan<- *core.TransactionEve return self.scope.Track(self.txPostFeed.Subscribe(ch)) } - // SubscribeLogsEvent implements the interface of filters.Backend // LightChain does not send logs events, so return an empty subscription. func (self *LightChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { diff --git a/rpc/subscription.go b/rpc/subscription.go index 279606c477..936cdde3ab 100644 --- a/rpc/subscription.go +++ b/rpc/subscription.go @@ -26,15 +26,15 @@ var ( // ErrNotificationsUnsupported is returned when the connection doesn't support notifications ErrNotificationsUnsupported = errors.New("notifications not supported") // ErrNotificationNotFound is returned when the notification for the given id is not found - ErrSubscriptionNotFound = errors.New("subscription not found") - ErrNamespaceNotFound = errors.New("namespace not found") + ErrSubscriptionNotFound = errors.New("subscription not found") + ErrNamespaceNotFound = errors.New("namespace not found") ErrSubscriptionTypeNotFound = errors.New("no active subscriptions of specified type") ) // ID defines a pseudo random number that is used to identify RPC subscriptions. 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 // filter types. type SubscriptionType byte @@ -53,31 +53,29 @@ const ( PendingTransactionsSubscription // BlocksSubscription queries hashes for blocks that are imported BlocksSubscription - // ReturnData queries for return data from transactions executed by a - // particular rpc client - ReturnDataSubscription + // ReturnData queries for return data from transactions executed by a + // particular rpc client + ReturnDataSubscription // LastSubscription keeps track of the last index LastIndexSubscription ) - - // 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(). type Subscription struct { ID ID namespace string - Type SubscriptionType - update chan interface{} // used to send update filter criteria of subscription - err chan error // closed on unsubscribe + Type SubscriptionType + update chan interface{} // used to send update filter criteria of subscription + err chan error // closed on unsubscribe } func (s *Subscription) SetType(subType SubscriptionType) { - s.Type = subType + s.Type = subType } 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. @@ -118,7 +116,7 @@ func NotifierFromContext(ctx context.Context) (*Notifier, bool) { // 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. 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.inactive[s.ID] = s n.subMu.Unlock() @@ -148,15 +146,15 @@ func (n *Notifier) Notify(id ID, data interface{}) error { // generate notifcations. // func (n *Notifier) UpdateSubscription(id ID, message interface{}) error { - n.subMu.RLock() - defer n.subMu.RUnlock() - sub, validid := n.active[id] - if validid { - sub.update <- message - return nil - } else { - return ErrSubscriptionNotFound - } + n.subMu.RLock() + defer n.subMu.RUnlock() + sub, validid := n.active[id] + if validid { + sub.update <- message + return nil + } else { + return ErrSubscriptionNotFound + } } // 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 // which should generate a TransactionEvent and notify the client with return data. func (n *Notifier) UpdateSubscriptions(subType SubscriptionType, message interface{}) error { - n.subMu.RLock() - defer n.subMu.RUnlock() - for _,sub := range n.active { - if sub.Type == subType { - sub.update <- message - return nil - } - } + n.subMu.RLock() + defer n.subMu.RUnlock() + for _, sub := range n.active { + if sub.Type == subType { + sub.update <- message + return nil + } + } - return ErrSubscriptionTypeNotFound + return ErrSubscriptionTypeNotFound } // Closed returns a channel that is closed when the RPC connection is closed.