diff --git a/core/blockchain.go b/core/blockchain.go index b33eb85a44..fc60463d0e 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -99,6 +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 scope event.SubscriptionScope genesisBlock *types.Block @@ -1371,8 +1372,8 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) { case ChainSideEvent: bc.chainSideFeed.Send(ev) - } - } + } + } } func (bc *BlockChain) update() { @@ -1558,6 +1559,11 @@ func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Su return bc.scope.Track(bc.chainSideFeed.Subscribe(ch)) } +// SubscribeTransactionEvent registers a subscription of SubscribeTransactionEvent. +func (bc *BlockChain) SubscribeTransactionEvent(ch chan<- *TransactionEvent) event.Subscription { + return bc.scope.Track(bc.txPostFeed.Subscribe(ch)) +} + // SubscribeLogsEvent registers a subscription of []*types.Log. func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { return bc.scope.Track(bc.logsFeed.Subscribe(ch)) diff --git a/core/events.go b/core/events.go index 6f404f612b..12723f0df8 100644 --- a/core/events.go +++ b/core/events.go @@ -35,6 +35,12 @@ type PendingStateEvent struct{} // NewMinedBlockEvent is posted when a block has been imported. type NewMinedBlockEvent struct{ Block *types.Block } +// TransactionEvent is posted when a transaction completes execution +type TransactionEvent struct{ + TxHash common.Hash + RetData types.ReturnData +} + // RemovedTransactionEvent is posted when a reorg happens type RemovedTransactionEvent struct{ Txs types.Transactions } diff --git a/core/state_processor.go b/core/state_processor.go index 4dc58b9dea..cf0fc494f0 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -96,10 +96,12 @@ func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common // about the transaction and calling mechanisms. vmenv := vm.NewEVM(context, statedb, config, cfg) // Apply the transaction to the current state (included in the env) - _, gas, failed, err := ApplyMessage(vmenv, msg, gp) + + data, gas, failed, err := ApplyMessage(vmenv, msg, gp) if err != nil { return nil, 0, err } + // Update the state with pending changes var root []byte if config.IsByzantium(header.Number) { @@ -122,5 +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) + return receipt, gas, err } diff --git a/core/types/transaction.go b/core/types/transaction.go index c1cb7a043a..7b387aa996 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -80,6 +80,18 @@ type txdataMarshaling struct { S *hexutil.Big } +// Return data generated by a transaction which calls a contract method +type ReturnData struct { + // 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"` +} + 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) } diff --git a/eth/api_backend.go b/eth/api_backend.go index ecd5488a24..d5ea49ce67 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -144,6 +144,10 @@ func (b *EthApiBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) e return b.eth.BlockChain().SubscribeChainSideEvent(ch) } +func (b *EthApiBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { + return b.eth.BlockChain().SubscribeTransactionEvent(ch) +} + func (b *EthApiBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { return b.eth.BlockChain().SubscribeLogsEvent(ch) } diff --git a/eth/filters/api.go b/eth/filters/api.go index ec403709c5..fbe2a88078 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -24,11 +24,13 @@ import ( "math/big" "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/rpc" @@ -46,6 +48,7 @@ type filter struct { hashes []common.Hash crit FilterCriteria logs []*types.Log + retdata []types.ReturnData s *Subscription // associated subscription in event system } @@ -229,6 +232,81 @@ func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, er return rpcSub, nil } +func (api *PublicFilterAPI) NewReturnDataFilter() rpc.ID { + var ( + retCh = make(chan *types.ReturnData) + retSub = api.events.SubscribeReturnData(retCh) + ) + + 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 { + select { + case retdata := <-retCh: + api.filtersMu.Lock() + if f, found := api.filters[retSub.ID]; found { + f.retdata = append(f.retdata, *retdata) + } + api.filtersMu.Unlock() + case <-retSub.Err(): + api.filtersMu.Lock() + delete(api.filters, retSub.ID) + api.filtersMu.Unlock() + return + } + } + }() + + return retSub.ID +} + +func (api *PublicFilterAPI) ReturnData(ctx context.Context) (*rpc.Subscription, error) { + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported + } + + txListen := make(map[common.Hash]bool) + + rpcSub := notifier.CreateSubscription() + rpcSub.SetType(rpc.ReturnDataSubscription) + + go func() { + retCh := make(chan *types.ReturnData) + retSub := api.events.SubscribeReturnData(retCh) + + 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(): + retSub.Unsubscribe() + return + case <-notifier.Closed(): + retSub.Unsubscribe() + return + } + } + }() + + return rpcSub, nil +} + + // Logs creates a subscription that fires for all new log that match the given filter criteria. func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) @@ -399,6 +477,7 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty // // 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() @@ -419,6 +498,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 } } @@ -443,6 +526,17 @@ func returnLogs(logs []*types.Log) []*types.Log { return logs } +func returnRetData(rdata []types.ReturnData) []types.ReturnData { + if rdata == nil { + return []types.ReturnData{} + } else { + for _,r := range rdata { + r.Data = []byte{} + } + } + return rdata +} + // UnmarshalJSON sets *args fields with given data. func (args *FilterCriteria) UnmarshalJSON(data []byte) error { type input struct { diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 5dfe60e772..964ff366a0 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -37,6 +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 SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription @@ -177,7 +178,7 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err } } -// indexedLogs returns the logs matching the filter criteria based on raw block +// unindexedLogs returns the logs matching the filter criteria based on raw block // iteration and bloom matching. func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) { var logs []*types.Log diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index f8097c7b95..78a648a6ac 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -35,31 +35,37 @@ import ( // Type determines the kind of filter and is used to put the filter in to // the correct bucket when added. -type Type byte +type Type = rpc.SubscriptionType const ( // UnknownSubscription indicates an unknown subscription type - UnknownSubscription Type = iota + UnknownSubscription Type = rpc.UnknownSubscription // LogsSubscription queries for new or removed (chain reorg) logs - LogsSubscription + LogsSubscription Type = rpc.LogsSubscription // PendingLogsSubscription queries for logs in pending blocks - PendingLogsSubscription + PendingLogsSubscription Type = rpc.PendingLogsSubscription // MinedAndPendingLogsSubscription queries for logs in mined and pending blocks. - MinedAndPendingLogsSubscription + MinedAndPendingLogsSubscription Type = rpc.MinedAndPendingLogsSubscription // PendingTransactionsSubscription queries tx hashes for pending // transactions entering the pending state - PendingTransactionsSubscription + PendingTransactionsSubscription Type = rpc.PendingTransactionsSubscription // BlocksSubscription queries hashes for blocks that are imported - BlocksSubscription + BlocksSubscription Type = rpc.BlocksSubscription + // 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 + LastIndexSubscription Type = rpc.LastIndexSubscription ) const ( - // txChanSize is the size of channel listening to TxPreEvent. + // txPreChanSize is the size of channel listening to TxPreEvent. // The number is referenced from the size of tx pool. - txChanSize = 4096 + txPreChanSize = 4096 + // txPostChanSize is the size of channel listening to TransactionEvent. + // For now, setting to same as txPreChanSize--good number? + txPostChanSize = 4096 // rmLogsChanSize is the size of channel listening to RemovedLogsEvent. rmLogsChanSize = 10 // logsChanSize is the size of channel listening to LogsEvent. @@ -73,15 +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 - 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 @@ -282,6 +289,20 @@ func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscr return es.subscribe(sub) } +// 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{ + id: rpc.NewID(), + typ: ReturnDataSubscription, + created: time.Now(), + retdata: retCh, + installed: make(chan struct{}), + err: make(chan error), + } + return es.subscribe(sub) +} + type filterIndex map[Type]map[rpc.ID]*subscription // broadcast event to filters that match criteria. @@ -320,6 +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.ChainEvent: for _, f := range filters[BlocksSubscription] { f.headers <- e.Block.Header() @@ -414,9 +439,12 @@ func (es *EventSystem) eventLoop() { var ( index = make(filterIndex) sub = es.mux.Subscribe(core.PendingLogsEvent{}) - // Subscribe TxPreEvent form txpool - txCh = make(chan core.TxPreEvent, txChanSize) - txSub = es.backend.SubscribeTxPreEvent(txCh) + // 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 RemovedLogsEvent rmLogsCh = make(chan core.RemovedLogsEvent, rmLogsChanSize) rmLogsSub = es.backend.SubscribeRemovedLogsEvent(rmLogsCh) @@ -430,6 +458,7 @@ func (es *EventSystem) eventLoop() { // Unsubscribe all events defer sub.Unsubscribe() + defer txPreSub.Unsubscribe() defer txSub.Unsubscribe() defer rmLogsSub.Unsubscribe() defer logsSub.Unsubscribe() @@ -448,6 +477,8 @@ func (es *EventSystem) eventLoop() { es.broadcast(index, ev) // Handle subscribed events + case ev := <-txPreCh: + es.broadcast(index, ev) case ev := <-txCh: es.broadcast(index, ev) case ev := <-rmLogsCh: @@ -477,6 +508,8 @@ func (es *EventSystem) eventLoop() { close(f.err) // System stopped + case <-txPreSub.Err(): + return case <-txSub.Err(): return case <-rmLogsSub.Err(): diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index ae7e252654..13b39c7caf 100644 --- a/ethstats/ethstats.go +++ b/ethstats/ethstats.go @@ -150,8 +150,8 @@ func (s *Service) loop() { headSub := blockchain.SubscribeChainHeadEvent(chainHeadCh) defer headSub.Unsubscribe() - txEventCh := make(chan core.TxPreEvent, txChanSize) - txSub := txpool.SubscribeTxPreEvent(txEventCh) + txPreEventCh := make(chan core.TxPreEvent, txChanSize) + txSub := txpool.SubscribeTxPreEvent(txPreEventCh) defer txSub.Unsubscribe() // Start a goroutine that exhausts the subsciptions to avoid events piling up @@ -174,7 +174,7 @@ func (s *Service) loop() { } // Notify of new transaction events, but drop if too frequent - case <-txEventCh: + case <-txPreEventCh: if time.Duration(mclock.Now()-lastTx) < time.Second { continue } diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index e2bfbaf307..e799b3f95b 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -1184,7 +1184,16 @@ func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (c } else { log.Info("Submitted transaction", "fullhash", tx.Hash().Hex(), "recipient", tx.To()) } - return tx.Hash(), nil + + 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 } // 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 af95d7906f..b9645f57b9 100644 --- a/internal/ethapi/backend.go +++ b/internal/ethapi/backend.go @@ -57,6 +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 // TxPool API SendTx(ctx context.Context, signedTx *types.Transaction) error diff --git a/les/api_backend.go b/les/api_backend.go index 3fc5c33a44..0cce15b142 100644 --- a/les/api_backend.go +++ b/les/api_backend.go @@ -133,6 +133,10 @@ func (b *LesApiBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Sub return b.eth.txPool.SubscribeTxPreEvent(ch) } +func (b *LesApiBackend) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { + return b.eth.blockchain.SubscribeTransactionEvent(ch) +} + func (b *LesApiBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription { return b.eth.blockchain.SubscribeChainEvent(ch) } diff --git a/light/lightchain.go b/light/lightchain.go index 2784615d35..eaa14195ca 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -52,6 +52,7 @@ type LightChain struct { chainFeed event.Feed chainSideFeed event.Feed chainHeadFeed event.Feed + txPostFeed event.Feed scope event.SubscriptionScope genesisBlock *types.Block @@ -488,6 +489,12 @@ func (self *LightChain) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) e return self.scope.Track(self.chainSideFeed.Subscribe(ch)) } +// SubscribeTransactionEvent registers a subscription of TransactionEvent. +func (self *LightChain) SubscribeTransactionEvent(ch chan<- *core.TransactionEvent) event.Subscription { + 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/server.go b/rpc/server.go index 0f29035edb..3ca4bcf708 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -145,7 +145,7 @@ func (s *Server) serveRequest(codec ServerCodec, singleShot bool, options CodecO defer cancel() // if the codec supports notification include a notifier that callbacks can use - // to send notification to clients. It is thight to the codec/connection. If the + // to send notification to clients. It is tied to the codec/connection. If the // connection is closed the notifier will stop and cancels all active subscriptions. if options&OptionSubscriptions == OptionSubscriptions { ctx = context.WithValue(ctx, notifierKey{}, newNotifier(codec)) @@ -283,7 +283,7 @@ func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverReque return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}), nil } - // active the subscription after the sub id was successfully sent to the client + // activate the subscription after the sub id was successfully sent to the client activateSub := func() { notifier, _ := NotifierFromContext(ctx) notifier.activate(subid, req.svcname) diff --git a/rpc/subscription.go b/rpc/subscription.go index 6ce7befa1d..279606c477 100644 --- a/rpc/subscription.go +++ b/rpc/subscription.go @@ -27,20 +27,60 @@ var ( 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") + 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 -// a Subscription is created by a notifier and tight to that notifier. The client can use +// 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 + +const ( + // UnknownSubscription indicates an unknown subscription type + UnknownSubscription SubscriptionType = iota + // LogsSubscription queries for new or removed (chain reorg) logs + LogsSubscription + // PendingLogsSubscription queries for logs in pending blocks + PendingLogsSubscription + // MinedAndPendingLogsSubscription queries for logs in mined and pending blocks. + MinedAndPendingLogsSubscription + // PendingTransactionsSubscription queries tx hashes for pending + // transactions entering the pending state + PendingTransactionsSubscription + // BlocksSubscription queries hashes for blocks that are imported + BlocksSubscription + // 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 } -// Err returns a channel that is closed when the client send an unsubscribe request. +func (s *Subscription) SetType(subType SubscriptionType) { + s.Type = subType +} + +func (s *Subscription) Update() <-chan interface{} { + return s.update +} + +// Err returns a channel that is closed when the client sends an unsubscribe request. func (s *Subscription) Err() <-chan error { return s.err } @@ -48,7 +88,7 @@ func (s *Subscription) Err() <-chan error { // notifierKey is used to store a notifier within the connection context. type notifierKey struct{} -// Notifier is tight to a RPC connection that supports subscriptions. +// Notifier is tied to a RPC connection that supports subscriptions. // Server callbacks use the notifier to send notifications. type Notifier struct { codec ServerCodec @@ -78,7 +118,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(), 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() @@ -102,6 +142,46 @@ func (n *Notifier) Notify(id ID, data interface{}) error { return nil } +// Send an update message to the update channel of a particular subscription id. +// +// Intended use pattern is to update the filter criteria about which events should +// 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 + } +} + +// Send an update message to the update channels of all subscriptions of +// the specified subscription type. +// +// Intended use pattern is to update the filter criteria about which events should +// generate notifcations. This version is especially useful if the subscription +// id is not known by the caller. Unlike UpdateScription, this can block. +// +// For example, when a client is subscribed to ReturnData, each time the client +// 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 + } + } + + return ErrSubscriptionTypeNotFound +} + // Closed returns a channel that is closed when the RPC connection is closed. func (n *Notifier) Closed() <-chan interface{} { return n.codec.Closed() diff --git a/rpc/utils.go b/rpc/utils.go index 9315cab591..220b956966 100644 --- a/rpc/utils.go +++ b/rpc/utils.go @@ -83,7 +83,7 @@ func isSubscriptionType(t reflect.Type) bool { return t == subscriptionType } -// isPubSub tests whether the given method has as as first argument a context.Context +// isPubSub tests whether the given method has a context.Context as a first argument // and returns the pair (Subscription, error) func isPubSub(methodType reflect.Type) bool { // numIn(0) is the receiver type