rpc,eth/filters,internal/ethapi,core: First draft of EIP758 implementation.

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))

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 }
 

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
 }

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)
 }

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)
 }

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 {

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

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():

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
 				}

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

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

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)
 }

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 {

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)

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()

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