fix: ethClient.SubscribeNewDeposit should accept StateData channel (#44)

* ethClient.SubscribeNewDeposit should accept StateData channel

* remove redundant function

* Fix compilation error
This commit is contained in:
Arpit Agarwal 2020-04-06 11:40:14 +05:30 committed by GitHub
parent 260243cf88
commit 7183e217c0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 17 additions and 15 deletions

View file

@ -510,6 +510,9 @@ func (fb *filterBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEve
func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { func (fb *filterBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
return fb.bc.SubscribeLogsEvent(ch) return fb.bc.SubscribeLogsEvent(ch)
} }
func (fb *filterBackend) SubscribeStateEvent(ch chan<- core.NewStateChangeEvent) event.Subscription {
return fb.bc.SubscribeStateEvent(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) {

View file

@ -1161,7 +1161,6 @@ func (c *Bor) CommitStates(
header *types.Header, header *types.Header,
chain core.ChainContext, chain core.ChainContext,
) error { ) error {
fmt.Println("comminting state")
// get pending state proposals // get pending state proposals
stateIds, err := c.GetPendingStateProposals(header.Number.Uint64() - 1) stateIds, err := c.GetPendingStateProposals(header.Number.Uint64() - 1)
if err != nil { if err != nil {
@ -1240,7 +1239,6 @@ func (c *Bor) CommitStates(
return nil return nil
} }
// SubscribeStateEvent registers a subscription of ChainSideEvent. // SubscribeStateEvent registers a subscription of ChainSideEvent.
func (c *Bor) SubscribeStateEvent(ch chan<- core.NewStateChangeEvent) event.Subscription { func (c *Bor) SubscribeStateEvent(ch chan<- core.NewStateChangeEvent) event.Subscription {
return c.scope.Track(c.stateDataFeed.Subscribe(ch)) return c.scope.Track(c.stateDataFeed.Subscribe(ch))

View file

@ -27,6 +27,7 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/hashicorp/golang-lru"
"github.com/maticnetwork/bor/common" "github.com/maticnetwork/bor/common"
"github.com/maticnetwork/bor/common/mclock" "github.com/maticnetwork/bor/common/mclock"
"github.com/maticnetwork/bor/common/prque" "github.com/maticnetwork/bor/common/prque"
@ -42,7 +43,6 @@ import (
"github.com/maticnetwork/bor/params" "github.com/maticnetwork/bor/params"
"github.com/maticnetwork/bor/rlp" "github.com/maticnetwork/bor/rlp"
"github.com/maticnetwork/bor/trie" "github.com/maticnetwork/bor/trie"
"github.com/hashicorp/golang-lru"
) )
var ( var (
@ -142,6 +142,7 @@ type BlockChain struct {
chainHeadFeed event.Feed chainHeadFeed event.Feed
logsFeed event.Feed logsFeed event.Feed
blockProcFeed event.Feed blockProcFeed event.Feed
stateDataFeed event.Feed
scope event.SubscriptionScope scope event.SubscriptionScope
genesisBlock *types.Block genesisBlock *types.Block
@ -2177,6 +2178,11 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
return bc.scope.Track(bc.logsFeed.Subscribe(ch)) return bc.scope.Track(bc.logsFeed.Subscribe(ch))
} }
// SubscribeStateEvent registers a subscription of ChainSideEvent.
func (bc *BlockChain) SubscribeStateEvent(ch chan<- NewStateChangeEvent) event.Subscription {
return bc.scope.Track(bc.stateDataFeed.Subscribe(ch))
}
// SubscribeBlockProcessingEvent registers a subscription of bool where true means // SubscribeBlockProcessingEvent registers a subscription of bool where true means
// block processing has started while false means it has stopped. // block processing has started while false means it has stopped.
func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription { func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {

View file

@ -79,10 +79,6 @@ type txdataMarshaling struct {
S *hexutil.Big S *hexutil.Big
} }
func (sd *StateData) StateData() *StateData {
return sd
}
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 {
return newTransaction(nonce, &to, amount, gasLimit, gasPrice, data) return newTransaction(nonce, &to, amount, gasLimit, gasPrice, data)
} }

View file

@ -17,6 +17,7 @@
package filters package filters
import ( import (
"bytes"
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
@ -247,7 +248,7 @@ func (api *PublicFilterAPI) NewDeposits(ctx context.Context, crit ethereum.Filte
for { for {
select { select {
case h := <-stateData: case h := <-stateData:
if crit.Did == h.Did || crit.Contract == h.Contract || if crit.Did == h.Did || bytes.Compare(crit.Contract.Bytes(), h.Contract.Bytes()) == 0 ||
(crit.Did == 0 && crit.Contract == common.Address{}) { (crit.Did == 0 && crit.Contract == common.Address{}) {
notifier.Notify(rpcSub.ID, h) notifier.Notify(rpcSub.ID, h)
} }

View file

@ -311,8 +311,7 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
return es.subscribe(sub) return es.subscribe(sub)
} }
// SubscribeNewHeads creates a subscription that writes the header of a block that is // SubscribeNewDeposits creates a subscription that writes details about the new state sync events (from mainchain to Bor)
// imported in the chain.
func (es *EventSystem) SubscribeNewDeposits(stateData chan *types.StateData) *Subscription { func (es *EventSystem) SubscribeNewDeposits(stateData chan *types.StateData) *Subscription {
sub := &subscription{ sub := &subscription{
id: rpc.NewID(), id: rpc.NewID(),
@ -387,7 +386,7 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
} }
case core.NewStateChangeEvent: case core.NewStateChangeEvent:
for _, f := range filters[StateSubscription] { for _, f := range filters[StateSubscription] {
f.stateData <- e.StateData.StateData() f.stateData <- e.StateData
} }
case core.ChainEvent: case core.ChainEvent:
for _, f := range filters[BlocksSubscription] { for _, f := range filters[BlocksSubscription] {

View file

@ -324,10 +324,9 @@ func (ec *Client) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header)
return ec.c.EthSubscribe(ctx, ch, "newHeads") return ec.c.EthSubscribe(ctx, ch, "newHeads")
} }
// SubscribeNewHead subscribes to notifications about the current blockchain head // SubscribeNewDeposit subscribes to new state sync events
// on the given channel. func (ec *Client) SubscribeNewDeposit(ctx context.Context, ch chan<- *types.StateData) (ethereum.Subscription, error) {
func (ec *Client) SubscribeNewDeposit(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) { return ec.c.EthSubscribe(ctx, ch, "newDeposits", nil)
return ec.c.EthSubscribe(ctx, ch, "newDeposits")
} }
// State Access // State Access