rpc: refactor subscriptions and filters

core/types/block.go

@@ -114,6 +114,28 @@ func (h *Header) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+func (h *Header) MarshalJSON() ([]byte, error) {
+	fields := map[string]interface{}{
+		"hash":             h.Hash(),
+		"parentHash":       h.ParentHash,
+		"number":           fmt.Sprintf("%#x", h.Number),
+		"nonce":            h.Nonce,
+		"receiptRoot":      h.ReceiptHash,
+		"logsBloom":        h.Bloom,
+		"sha3Uncles":       h.UncleHash,
+		"stateRoot":        h.Root,
+		"miner":            h.Coinbase,
+		"difficulty":       fmt.Sprintf("%#x", h.Difficulty),
+		"extraData":        fmt.Sprintf("0x%x", h.Extra),
+		"gasLimit":         fmt.Sprintf("%#x", h.GasLimit),
+		"gasUsed":          fmt.Sprintf("%#x", h.GasUsed),
+		"timestamp":        fmt.Sprintf("%#x", h.Time),
+		"transactionsRoot": h.TxHash,
+	}
+
+	return json.Marshal(fields)
+}
+
 func rlpHash(x interface{}) (h common.Hash) {
 	hw := sha3.NewKeccak256()
 	rlp.Encode(hw, x)

eth/downloader/api.go

@@ -19,53 +19,102 @@ package downloader
 import (
 	"sync"
 
-	"golang.org/x/net/context"
-
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/rpc"
+	"golang.org/x/net/context"
 )
 
 // PublicDownloaderAPI provides an API which gives information about the current synchronisation status.
 // It offers only methods that operates on data that can be available to anyone without security risks.
 type PublicDownloaderAPI struct {
-	d                   *Downloader
-	mux                 *event.TypeMux
-	muSyncSubscriptions sync.Mutex
-	syncSubscriptions   map[string]rpc.Subscription
+	d                         *Downloader
+	mux                       *event.TypeMux
+	installSyncSubscription   chan chan interface{}
+	uninstallSyncSubscription chan *uninstallSyncSubscriptionRequest
 }
 
-// NewPublicDownloaderAPI create a new PublicDownloaderAPI.
+// NewPublicDownloaderAPI create a new PublicDownloaderAPI. The API has an internal event loop that
+// listens for events from the downloader through the global event mux. In case it receives one of
+// these events it broadcasts it to all syncing subscriptions that are installed through the
+// installSyncSubscription channel.
 func NewPublicDownloaderAPI(d *Downloader, m *event.TypeMux) *PublicDownloaderAPI {
-	api := &PublicDownloaderAPI{d: d, mux: m, syncSubscriptions: make(map[string]rpc.Subscription)}
+	api := &PublicDownloaderAPI{
+		d:   d,
+		mux: m,
+		installSyncSubscription:   make(chan chan interface{}),
+		uninstallSyncSubscription: make(chan *uninstallSyncSubscriptionRequest),
+	}
 
-	go api.run()
+	go api.eventLoop()
 
 	return api
 }
 
-func (api *PublicDownloaderAPI) run() {
-	sub := api.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{})
-
-	for event := range sub.Chan() {
-		var notification interface{}
+// eventLoop runs an loop until the event mux closes. It will install and uninstall new
+// sync subscriptions and broadcasts sync status updates to the installed sync subscriptions.
+func (api *PublicDownloaderAPI) eventLoop() {
+	var (
+		sub               = api.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{})
+		syncSubscriptions = make(map[chan interface{}]struct{})
+	)
+
+	for {
+		select {
+		case i := <-api.installSyncSubscription:
+			syncSubscriptions[i] = struct{}{}
+		case u := <-api.uninstallSyncSubscription:
+			delete(syncSubscriptions, u.c)
+			close(u.uninstalled)
+		case event := <-sub.Chan():
+			if event == nil {
+				return
+			}
 
-		switch event.Data.(type) {
-		case StartEvent:
-			result := &SyncingResult{Syncing: true}
-			result.Status.Origin, result.Status.Current, result.Status.Height, result.Status.Pulled, result.Status.Known = api.d.Progress()
-			notification = result
-		case DoneEvent, FailedEvent:
-			notification = false
+			var notification interface{}
+			switch event.Data.(type) {
+			case StartEvent:
+				result := &SyncingResult{Syncing: true}
+				result.Status.Origin, result.Status.Current, result.Status.Height, result.Status.Pulled, result.Status.Known = api.d.Progress()
+				notification = result
+			case DoneEvent, FailedEvent:
+				notification = false
+			}
+			// broadcast
+			for c := range syncSubscriptions {
+				c <- notification
+			}
 		}
+	}
+}
+
+// Syncing provides information when this nodes starts synchronising with the Ethereum network and when it's finished.
+func (api *PublicDownloaderAPI) Syncing(ctx context.Context) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
+	}
 
-		api.muSyncSubscriptions.Lock()
-		for id, sub := range api.syncSubscriptions {
-			if sub.Notify(notification) == rpc.ErrNotificationNotFound {
-				delete(api.syncSubscriptions, id)
+	rpcSub := notifier.CreateSubscription()
+
+	go func() {
+		statuses := make(chan interface{})
+		sub := api.SubscribeSyncStatus(statuses)
+
+		for {
+			select {
+			case status := <-statuses:
+				notifier.Notify(rpcSub.ID, status)
+			case <-rpcSub.Err():
+				sub.Unsubscribe()
+				return
+			case <-notifier.Closed():
+				sub.Unsubscribe()
+				return
 			}
 		}
-		api.muSyncSubscriptions.Unlock()
-	}
+	}()
+
+	return rpcSub, nil
 }
 
 // Progress gives progress indications when the node is synchronising with the Ethereum network.
@@ -83,26 +132,42 @@ type SyncingResult struct {
 	Status  Progress `json:"status"`
 }
 
-// Syncing provides information when this nodes starts synchronising with the Ethereum network and when it's finished.
-func (api *PublicDownloaderAPI) Syncing(ctx context.Context) (rpc.Subscription, error) {
-	notifier, supported := rpc.NotifierFromContext(ctx)
-	if !supported {
-		return nil, rpc.ErrNotificationsUnsupported
-	}
-
-	subscription, err := notifier.NewSubscription(func(id string) {
-		api.muSyncSubscriptions.Lock()
-		delete(api.syncSubscriptions, id)
-		api.muSyncSubscriptions.Unlock()
-	})
+// uninstallSyncSubscriptionRequest uninstalles a syncing subscription in the API event loop.
+type uninstallSyncSubscriptionRequest struct {
+	c           chan interface{}
+	uninstalled chan interface{}
+}
 
-	if err != nil {
-		return nil, err
-	}
+// SyncStatusSubscription represents a syncing subscription.
+type SyncStatusSubscription struct {
+	api       *PublicDownloaderAPI // register subscription in event loop of this api instance
+	c         chan interface{}     // channel where events are broadcasted to
+	unsubOnce sync.Once            // make sure unsubscribe logic is executed once
+}
 
-	api.muSyncSubscriptions.Lock()
-	api.syncSubscriptions[subscription.ID()] = subscription
-	api.muSyncSubscriptions.Unlock()
+// Unsubscribe uninstalls the subscription from the DownloadAPI event loop.
+// The status channel that was passed to subscribeSyncStatus isn't used anymore
+// after this method returns.
+func (s *SyncStatusSubscription) Unsubscribe() {
+	s.unsubOnce.Do(func() {
+		req := uninstallSyncSubscriptionRequest{s.c, make(chan interface{})}
+		s.api.uninstallSyncSubscription <- &req
+
+		for {
+			select {
+			case <-s.c:
+				// drop new status events until uninstall confirmation
+				continue
+			case <-req.uninstalled:
+				return
+			}
+		}
+	})
+}
 
-	return subscription, nil
+// SubscribeSyncStatus creates a subscription that will broadcast new synchronisation updates.
+// The given channel must receive interface values, the result can either
+func (api *PublicDownloaderAPI) SubscribeSyncStatus(status chan interface{}) *SyncStatusSubscription {
+	api.installSyncSubscription <- status
+	return &SyncStatusSubscription{api: api, c: status}
 }

eth/filters/api_test.go

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
 
-package filters_test
+package filters
 
 import (
 	"encoding/json"
@@ -22,7 +22,6 @@ import (
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/eth/filters"
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
@@ -39,14 +38,14 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	)
 
 	// default values
-	var test0 filters.NewFilterArgs
+	var test0 FilterCriteria
 	if err := json.Unmarshal([]byte("{}"), &test0); err != nil {
 		t.Fatal(err)
 	}
-	if test0.FromBlock != rpc.LatestBlockNumber {
+	if test0.FromBlock.Int64() != rpc.LatestBlockNumber.Int64() {
 		t.Fatalf("expected %d, got %d", rpc.LatestBlockNumber, test0.FromBlock)
 	}
-	if test0.ToBlock != rpc.LatestBlockNumber {
+	if test0.ToBlock.Int64() != rpc.LatestBlockNumber.Int64() {
 		t.Fatalf("expected %d, got %d", rpc.LatestBlockNumber, test0.ToBlock)
 	}
 	if len(test0.Addresses) != 0 {
@@ -57,20 +56,20 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	}
 
 	// from, to block number
-	var test1 filters.NewFilterArgs
+	var test1 FilterCriteria
 	vector := fmt.Sprintf(`{"fromBlock":"0x%x","toBlock":"0x%x"}`, fromBlock, toBlock)
 	if err := json.Unmarshal([]byte(vector), &test1); err != nil {
 		t.Fatal(err)
 	}
-	if test1.FromBlock != fromBlock {
+	if test1.FromBlock.Int64() != fromBlock.Int64() {
 		t.Fatalf("expected FromBlock %d, got %d", fromBlock, test1.FromBlock)
 	}
-	if test1.ToBlock != toBlock {
+	if test1.ToBlock.Int64() != toBlock.Int64() {
 		t.Fatalf("expected ToBlock %d, got %d", toBlock, test1.ToBlock)
 	}
 
 	// single address
-	var test2 filters.NewFilterArgs
+	var test2 FilterCriteria
 	vector = fmt.Sprintf(`{"address": "%s"}`, address0.Hex())
 	if err := json.Unmarshal([]byte(vector), &test2); err != nil {
 		t.Fatal(err)
@@ -83,7 +82,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	}
 
 	// multiple address
-	var test3 filters.NewFilterArgs
+	var test3 FilterCriteria
 	vector = fmt.Sprintf(`{"address": ["%s", "%s"]}`, address0.Hex(), address1.Hex())
 	if err := json.Unmarshal([]byte(vector), &test3); err != nil {
 		t.Fatal(err)
@@ -99,7 +98,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	}
 
 	// single topic
-	var test4 filters.NewFilterArgs
+	var test4 FilterCriteria
 	vector = fmt.Sprintf(`{"topics": ["%s"]}`, topic0.Hex())
 	if err := json.Unmarshal([]byte(vector), &test4); err != nil {
 		t.Fatal(err)
@@ -115,7 +114,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	}
 
 	// test multiple "AND" topics
-	var test5 filters.NewFilterArgs
+	var test5 FilterCriteria
 	vector = fmt.Sprintf(`{"topics": ["%s", "%s"]}`, topic0.Hex(), topic1.Hex())
 	if err := json.Unmarshal([]byte(vector), &test5); err != nil {
 		t.Fatal(err)
@@ -137,7 +136,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	}
 
 	// test optional topic
-	var test6 filters.NewFilterArgs
+	var test6 FilterCriteria
 	vector = fmt.Sprintf(`{"topics": ["%s", null, "%s"]}`, topic0.Hex(), topic2.Hex())
 	if err := json.Unmarshal([]byte(vector), &test6); err != nil {
 		t.Fatal(err)
@@ -165,7 +164,7 @@ func TestUnmarshalJSONNewFilterArgs(t *testing.T) {
 	}
 
 	// test OR topics
-	var test7 filters.NewFilterArgs
+	var test7 FilterCriteria
 	vector = fmt.Sprintf(`{"topics": [["%s", "%s"], null, ["%s", null]]}`, topic0.Hex(), topic1.Hex(), topic2.Hex())
 	if err := json.Unmarshal([]byte(vector), &test7); err != nil {
 		t.Fatal(err)

eth/filters/filter.go

@@ -23,15 +23,10 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/ethdb"
 )
 
-type AccountChange struct {
-	Address, StateAddress []byte
-}
-
-// Filtering interface
+// Filter can be used to retrieve and filter logs
 type Filter struct {
 	created time.Time
 
@@ -39,70 +34,72 @@ type Filter struct {
 	begin, end int64
 	addresses  []common.Address
 	topics     [][]common.Hash
-
-	BlockCallback       func(*types.Block, vm.Logs)
-	TransactionCallback func(*types.Transaction)
-	LogCallback         func(*vm.Log, bool)
 }
 
-// Create a new filter which uses a bloom filter on blocks to figure out whether a particular block
-// is interesting or not.
+// New creates a new filter which uses a bloom filter on blocks to figure out whether
+// a particular block is interesting or not.
 func New(db ethdb.Database) *Filter {
 	return &Filter{db: db}
 }
 
-// Set the earliest and latest block for filtering.
+// SetBeginBlock sets the earliest block for filtering.
 // -1 = latest block (i.e., the current block)
 // hash = particular hash from-to
-func (self *Filter) SetBeginBlock(begin int64) {
-	self.begin = begin
+func (f *Filter) SetBeginBlock(begin int64) {
+	f.begin = begin
 }
 
-func (self *Filter) SetEndBlock(end int64) {
-	self.end = end
+// SetEndBlock sets the latest block for filtering.
+func (f *Filter) SetEndBlock(end int64) {
+	f.end = end
 }
 
-func (self *Filter) SetAddresses(addr []common.Address) {
-	self.addresses = addr
+// SetAddresses matches only logs that are generated from addresses that are included
+// in the given addresses.
+func (f *Filter) SetAddresses(addr []common.Address) {
+	f.addresses = addr
 }
 
-func (self *Filter) SetTopics(topics [][]common.Hash) {
-	self.topics = topics
+// SetTopics matches only logs that have topics matching the given topics.
+func (f *Filter) SetTopics(topics [][]common.Hash) {
+	f.topics = topics
 }
 
 // Run filters logs with the current parameters set
-func (self *Filter) Find() vm.Logs {
-	latestHash := core.GetHeadBlockHash(self.db)
-	latestBlock := core.GetBlock(self.db, latestHash, core.GetBlockNumber(self.db, latestHash))
+func (f *Filter) Find() []Log {
+	latestHash := core.GetHeadBlockHash(f.db)
+	latestBlock := core.GetBlock(f.db, latestHash, core.GetBlockNumber(f.db, latestHash))
 	if latestBlock == nil {
-		return vm.Logs{}
+		return []Log{}
 	}
-	var beginBlockNo uint64 = uint64(self.begin)
-	if self.begin == -1 {
+
+	var beginBlockNo uint64 = uint64(f.begin)
+	if f.begin == -1 {
 		beginBlockNo = latestBlock.NumberU64()
 	}
-	var endBlockNo uint64 = uint64(self.end)
-	if self.end == -1 {
+
+	endBlockNo := uint64(f.end)
+	if f.end == -1 {
 		endBlockNo = latestBlock.NumberU64()
 	}
 
 	// if no addresses are present we can't make use of fast search which
 	// uses the mipmap bloom filters to check for fast inclusion and uses
 	// higher range probability in order to ensure at least a false positive
-	if len(self.addresses) == 0 {
-		return self.getLogs(beginBlockNo, endBlockNo)
+	if len(f.addresses) == 0 {
+		return f.getLogs(beginBlockNo, endBlockNo)
 	}
-	return self.mipFind(beginBlockNo, endBlockNo, 0)
+	return f.mipFind(beginBlockNo, endBlockNo, 0)
 }
 
-func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) {
+func (f *Filter) mipFind(start, end uint64, depth int) (logs []Log) {
 	level := core.MIPMapLevels[depth]
 	// normalise numerator so we can work in level specific batches and
 	// work with the proper range checks
 	for num := start / level * level; num <= end; num += level {
 		// find addresses in bloom filters
-		bloom := core.GetMipmapBloom(self.db, num, level)
-		for _, addr := range self.addresses {
+		bloom := core.GetMipmapBloom(f.db, num, level)
+		for _, addr := range f.addresses {
 			if bloom.TestBytes(addr[:]) {
 				// range check normalised values and make sure that
 				// we're resolving the correct range instead of the
@@ -110,9 +107,9 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) {
 				start := uint64(math.Max(float64(num), float64(start)))
 				end := uint64(math.Min(float64(num+level-1), float64(end)))
 				if depth+1 == len(core.MIPMapLevels) {
-					logs = append(logs, self.getLogs(start, end)...)
+					logs = append(logs, f.getLogs(start, end)...)
 				} else {
-					logs = append(logs, self.mipFind(start, end, depth+1)...)
+					logs = append(logs, f.mipFind(start, end, depth+1)...)
 				}
 				// break so we don't check the same range for each
 				// possible address. Checks on multiple addresses
@@ -125,12 +122,15 @@ func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) {
 	return logs
 }
 
-func (self *Filter) getLogs(start, end uint64) (logs vm.Logs) {
+func (f *Filter) getLogs(start, end uint64) (logs []Log) {
+	var block *types.Block
+
 	for i := start; i <= end; i++ {
-		var block *types.Block
-		hash := core.GetCanonicalHash(self.db, i)
+		hash := core.GetCanonicalHash(f.db, i)
 		if hash != (common.Hash{}) {
-			block = core.GetBlock(self.db, hash, i)
+			block = core.GetBlock(f.db, hash, i)
+		} else { // block not found
+			return logs
 		}
 		if block == nil { // block not found/written
 			return logs
@@ -138,16 +138,20 @@ func (self *Filter) getLogs(start, end uint64) (logs vm.Logs) {
 
 		// Use bloom filtering to see if this block is interesting given the
 		// current parameters
-		if self.bloomFilter(block) {
+		if f.bloomFilter(block) {
 			// Get the logs of the block
 			var (
-				receipts   = core.GetBlockReceipts(self.db, block.Hash(), i)
-				unfiltered vm.Logs
+				receipts   = core.GetBlockReceipts(f.db, block.Hash(), i)
+				unfiltered []Log
 			)
 			for _, receipt := range receipts {
-				unfiltered = append(unfiltered, receipt.Logs...)
+				rl := make([]Log, len(receipt.Logs))
+				for i, l := range receipt.Logs {
+					rl[i] = Log{l, false}
+				}
+				unfiltered = append(unfiltered, rl...)
 			}
-			logs = append(logs, self.FilterLogs(unfiltered)...)
+			logs = append(logs, filterLogs(unfiltered, f.addresses, f.topics)...)
 		}
 	}
 
@@ -164,26 +168,25 @@ func includes(addresses []common.Address, a common.Address) bool {
 	return false
 }
 
-func (self *Filter) FilterLogs(logs vm.Logs) vm.Logs {
-	var ret vm.Logs
+func filterLogs(logs []Log, addresses []common.Address, topics [][]common.Hash) []Log {
+	var ret []Log
 
 	// Filter the logs for interesting stuff
 Logs:
 	for _, log := range logs {
-		if len(self.addresses) > 0 && !includes(self.addresses, log.Address) {
+		if len(addresses) > 0 && !includes(addresses, log.Address) {
 			continue
 		}
 
-		logTopics := make([]common.Hash, len(self.topics))
+		logTopics := make([]common.Hash, len(topics))
 		copy(logTopics, log.Topics)
 
-		// If the to filtered topics is greater than the amount of topics in
-		//  logs, skip.
-		if len(self.topics) > len(log.Topics) {
+		// If the to filtered topics is greater than the amount of topics in logs, skip.
+		if len(topics) > len(log.Topics) {
 			continue Logs
 		}
 
-		for i, topics := range self.topics {
+		for i, topics := range topics {
 			var match bool
 			for _, topic := range topics {
 				// common.Hash{} is a match all (wildcard)
@@ -196,7 +199,6 @@ Logs:
 			if !match {
 				continue Logs
 			}
-
 		}
 
 		ret = append(ret, log)
@@ -205,10 +207,10 @@ Logs:
 	return ret
 }
 
-func (self *Filter) bloomFilter(block *types.Block) bool {
-	if len(self.addresses) > 0 {
+func (f *Filter) bloomFilter(block *types.Block) bool {
+	if len(f.addresses) > 0 {
 		var included bool
-		for _, addr := range self.addresses {
+		for _, addr := range f.addresses {
 			if types.BloomLookup(block.Bloom(), addr) {
 				included = true
 				break
@@ -220,7 +222,7 @@ func (self *Filter) bloomFilter(block *types.Block) bool {
 		}
 	}
 
-	for _, sub := range self.topics {
+	for _, sub := range f.topics {
 		var included bool
 		for _, topic := range sub {
 			if (topic == common.Hash{}) || types.BloomLookup(block.Bloom(), topic) {

internal/ethapi/api.go

@@ -24,7 +24,6 @@ import (
 	"fmt"
 	"math/big"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/ethereum/ethash"
@@ -345,37 +344,12 @@ func (s *PrivateAccountAPI) SignAndSendTransaction(ctx context.Context, args Sen
 // PublicBlockChainAPI provides an API to access the Ethereum blockchain.
 // It offers only methods that operate on public data that is freely available to anyone.
 type PublicBlockChainAPI struct {
-	b                       Backend
-	muNewBlockSubscriptions sync.Mutex                             // protects newBlocksSubscriptions
-	newBlockSubscriptions   map[string]func(core.ChainEvent) error // callbacks for new block subscriptions
+	b Backend
 }
 
 // NewPublicBlockChainAPI creates a new Etheruem blockchain API.
 func NewPublicBlockChainAPI(b Backend) *PublicBlockChainAPI {
-	api := &PublicBlockChainAPI{
-		b: b,
-		newBlockSubscriptions: make(map[string]func(core.ChainEvent) error),
-	}
-
-	go api.subscriptionLoop()
-
-	return api
-}
-
-// subscriptionLoop reads events from the global event mux and creates notifications for the matched subscriptions.
-func (s *PublicBlockChainAPI) subscriptionLoop() {
-	sub := s.b.EventMux().Subscribe(core.ChainEvent{})
-	for event := range sub.Chan() {
-		if chainEvent, ok := event.Data.(core.ChainEvent); ok {
-			s.muNewBlockSubscriptions.Lock()
-			for id, notifyOf := range s.newBlockSubscriptions {
-				if notifyOf(chainEvent) == rpc.ErrNotificationNotFound {
-					delete(s.newBlockSubscriptions, id)
-				}
-			}
-			s.muNewBlockSubscriptions.Unlock()
-		}
-	}
+	return &PublicBlockChainAPI{b}
 }
 
 // BlockNumber returns the block number of the chain head.
@@ -470,45 +444,6 @@ func (s *PublicBlockChainAPI) GetUncleCountByBlockHash(ctx context.Context, bloc
 	return nil
 }
 
-// NewBlocksArgs allows the user to specify if the returned block should include transactions and in which format.
-type NewBlocksArgs struct {
-	IncludeTransactions bool `json:"includeTransactions"`
-	TransactionDetails  bool `json:"transactionDetails"`
-}
-
-// NewBlocks triggers a new block event each time a block is appended to the chain. It accepts an argument which allows
-// the caller to specify whether the output should contain transactions and in what format.
-func (s *PublicBlockChainAPI) NewBlocks(ctx context.Context, args NewBlocksArgs) (rpc.Subscription, error) {
-	notifier, supported := rpc.NotifierFromContext(ctx)
-	if !supported {
-		return nil, rpc.ErrNotificationsUnsupported
-	}
-
-	// create a subscription that will remove itself when unsubscribed/cancelled
-	subscription, err := notifier.NewSubscription(func(subId string) {
-		s.muNewBlockSubscriptions.Lock()
-		delete(s.newBlockSubscriptions, subId)
-		s.muNewBlockSubscriptions.Unlock()
-	})
-
-	if err != nil {
-		return nil, err
-	}
-
-	// add a callback that is called on chain events which will format the block and notify the client
-	s.muNewBlockSubscriptions.Lock()
-	s.newBlockSubscriptions[subscription.ID()] = func(e core.ChainEvent) error {
-		notification, err := s.rpcOutputBlock(e.Block, args.IncludeTransactions, args.TransactionDetails)
-		if err == nil {
-			return subscription.Notify(notification)
-		}
-		glog.V(logger.Warn).Info("unable to format block %v\n", err)
-		return nil
-	}
-	s.muNewBlockSubscriptions.Unlock()
-	return subscription, nil
-}
-
 // GetCode returns the code stored at the given address in the state for the given block number.
 func (s *PublicBlockChainAPI) GetCode(ctx context.Context, address common.Address, blockNr rpc.BlockNumber) (string, error) {
 	state, _, err := s.b.StateAndHeaderByNumber(blockNr)
@@ -867,40 +802,12 @@ func newRPCTransaction(b *types.Block, txHash common.Hash) (*RPCTransaction, err
 
 // PublicTransactionPoolAPI exposes methods for the RPC interface
 type PublicTransactionPoolAPI struct {
-	b               Backend
-	muPendingTxSubs sync.Mutex
-	pendingTxSubs   map[string]rpc.Subscription
+	b Backend
 }
 
 // NewPublicTransactionPoolAPI creates a new RPC service with methods specific for the transaction pool.
 func NewPublicTransactionPoolAPI(b Backend) *PublicTransactionPoolAPI {
-	api := &PublicTransactionPoolAPI{
-		b:             b,
-		pendingTxSubs: make(map[string]rpc.Subscription),
-	}
-
-	go api.subscriptionLoop()
-
-	return api
-}
-
-// subscriptionLoop listens for events on the global event mux and creates notifications for subscriptions.
-func (s *PublicTransactionPoolAPI) subscriptionLoop() {
-	sub := s.b.EventMux().Subscribe(core.TxPreEvent{})
-	for event := range sub.Chan() {
-		tx := event.Data.(core.TxPreEvent)
-		if from, err := tx.Tx.FromFrontier(); err == nil {
-			if s.b.AccountManager().HasAddress(from) {
-				s.muPendingTxSubs.Lock()
-				for id, sub := range s.pendingTxSubs {
-					if sub.Notify(tx.Tx.Hash()) == rpc.ErrNotificationNotFound {
-						delete(s.pendingTxSubs, id)
-					}
-				}
-				s.muPendingTxSubs.Unlock()
-			}
-		}
-	}
+	return &PublicTransactionPoolAPI{b}
 }
 
 func getTransaction(chainDb ethdb.Database, b Backend, txHash common.Hash) (*types.Transaction, bool, error) {
@@ -1353,31 +1260,6 @@ func (s *PublicTransactionPoolAPI) PendingTransactions() []*RPCTransaction {
 	return transactions
 }
 
-// NewPendingTransactions creates a subscription that is triggered each time a transaction enters the transaction pool
-// and is send from one of the transactions this nodes manages.
-func (s *PublicTransactionPoolAPI) NewPendingTransactions(ctx context.Context) (rpc.Subscription, error) {
-	notifier, supported := rpc.NotifierFromContext(ctx)
-	if !supported {
-		return nil, rpc.ErrNotificationsUnsupported
-	}
-
-	subscription, err := notifier.NewSubscription(func(id string) {
-		s.muPendingTxSubs.Lock()
-		delete(s.pendingTxSubs, id)
-		s.muPendingTxSubs.Unlock()
-	})
-
-	if err != nil {
-		return nil, err
-	}
-
-	s.muPendingTxSubs.Lock()
-	s.pendingTxSubs[subscription.ID()] = subscription
-	s.muPendingTxSubs.Unlock()
-
-	return subscription, nil
-}
-
 // Resend accepts an existing transaction and a new gas price and limit. It will remove the given transaction from the
 // pool and reinsert it with the new gas price and limit.
 func (s *PublicTransactionPoolAPI) Resend(ctx context.Context, tx *Tx, gasPrice, gasLimit *rpc.HexNumber) (common.Hash, error) {

rpc/server.go

@@ -166,7 +166,7 @@ func (s *Server) serveRequest(codec ServerCodec, singleShot bool, options CodecO
 	// to send notification to clients. It is thight 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{}, newBufferedNotifier(codec, notificationBufferSize))
+		ctx = context.WithValue(ctx, notifierKey{}, newNotifier(codec))
 	}
 	s.codecsMu.Lock()
 	if atomic.LoadInt32(&s.run) != 1 { // server stopped
@@ -247,7 +247,7 @@ func (s *Server) Stop() {
 }
 
 // createSubscription will call the subscription callback and returns the subscription id or error.
-func (s *Server) createSubscription(ctx context.Context, c ServerCodec, req *serverRequest) (string, error) {
+func (s *Server) createSubscription(ctx context.Context, c ServerCodec, req *serverRequest) (ID, error) {
 	// subscription have as first argument the context following optional arguments
 	args := []reflect.Value{req.callb.rcvr, reflect.ValueOf(ctx)}
 	args = append(args, req.args...)
@@ -257,7 +257,7 @@ func (s *Server) createSubscription(ctx context.Context, c ServerCodec, req *ser
 		return "", reply[1].Interface().(error)
 	}
 
-	return reply[0].Interface().(Subscription).ID(), nil
+	return reply[0].Interface().(*Subscription).ID, nil
 }
 
 // handle executes a request and returns the response from the callback.
@@ -273,8 +273,8 @@ func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverReque
 				return codec.CreateErrorResponse(&req.id, &callbackError{ErrNotificationsUnsupported.Error()}), nil
 			}
 
-			subid := req.args[0].String()
-			if err := notifier.Unsubscribe(subid); err != nil {
+			subid := ID(req.args[0].String())
+			if err := notifier.unsubscribe(subid); err != nil {
 				return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}), nil
 			}
 
@@ -292,7 +292,7 @@ func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverReque
 		// active the subscription after the sub id was successfully sent to the client
 		activateSub := func() {
 			notifier, _ := NotifierFromContext(ctx)
-			notifier.(*bufferedNotifier).activate(subid)
+			notifier.activate(subid)
 		}
 
 		return codec.CreateResponse(req.id, subid), activateSub

rpc/server_test.go

@@ -72,7 +72,7 @@ func (s *Service) InvalidRets3() (string, string, error) {
 	return "", "", nil
 }
 
-func (s *Service) Subscription(ctx context.Context) (Subscription, error) {
+func (s *Service) Subscription(ctx context.Context) (*Subscription, error) {
 	return nil, nil
 }
 

rpc/subscription.go

+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package rpc
+
+import (
+	"errors"
+	"sync"
+
+	"golang.org/x/net/context"
+)
+
+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")
+)
+
+// ID defines a psuedo 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
+// this subscription to wait for an unsubscribe request for the client, see Err().
+type Subscription struct {
+	ID  ID
+	err chan error // closed on unsubscribe
+}
+
+// Err returns a channel that is closed when the client send an unsubscribe request.
+func (s *Subscription) Err() <-chan error {
+	return s.err
+}
+
+// notifierKey is used to store a notifier within the connection context.
+type notifierKey struct{}
+
+// Notifier is tight to a RPC connection that supports subscriptions.
+// Server callbacks use the notifier to send notifications.
+type Notifier struct {
+	codec    ServerCodec
+	subMu    sync.RWMutex // guards active and inactive maps
+	stopped  bool
+	active   map[ID]*Subscription
+	inactive map[ID]*Subscription
+}
+
+// newNotifier creates a new notifier that can be used to send subscription
+// notifications to the client.
+func newNotifier(codec ServerCodec) *Notifier {
+	return &Notifier{
+		codec:    codec,
+		active:   make(map[ID]*Subscription),
+		inactive: make(map[ID]*Subscription),
+	}
+}
+
+// NotifierFromContext returns the Notifier value stored in ctx, if any.
+func NotifierFromContext(ctx context.Context) (*Notifier, bool) {
+	n, ok := ctx.Value(notifierKey{}).(*Notifier)
+	return n, ok
+}
+
+// CreateSubscription returns a new subscription that is coupled to the
+// RPC connection. By default subscriptions are inactive and notifications
+// 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{NewID(), make(chan error)}
+	n.subMu.Lock()
+	n.inactive[s.ID] = s
+	n.subMu.Unlock()
+	return s
+}
+
+// Notify sends a notification to the client with the given data as payload.
+// If an error occurs the RPC connection is closed and the error is returned.
+func (n *Notifier) Notify(id ID, data interface{}) error {
+	n.subMu.RLock()
+	defer n.subMu.RUnlock()
+
+	_, active := n.active[id]
+	if active {
+		notification := n.codec.CreateNotification(string(id), data)
+		if err := n.codec.Write(notification); err != nil {
+			n.codec.Close()
+			return err
+		}
+	}
+	return nil
+}
+
+// Closed returns a channel that is closed when the RPC connection is closed.
+func (n *Notifier) Closed() <-chan interface{} {
+	return n.codec.Closed()
+}
+
+// unsubscribe a subscription.
+// If the subscription could not be found ErrSubscriptionNotFound is returned.
+func (n *Notifier) unsubscribe(id ID) error {
+	n.subMu.Lock()
+	defer n.subMu.Unlock()
+	if s, found := n.active[id]; found {
+		close(s.err)
+		delete(n.active, id)
+		return nil
+	}
+	return ErrSubscriptionNotFound
+}
+
+// activate enables a subscription. Until a subscription is enabled all
+// notifications are dropped. This method is called by the RPC server after
+// the subscription ID was sent to client. This prevents notifications being
+// send to the client before the subscription ID is send to the client.
+func (n *Notifier) activate(id ID) {
+	n.subMu.Lock()
+	defer n.subMu.Unlock()
+	if sub, found := n.inactive[id]; found {
+		n.active[id] = sub
+		delete(n.inactive, id)
+	}
+}

rpc/notification_test.go → rpc/subscription_test.go

@@ -50,7 +50,7 @@ func (s *NotificationTestService) Unsubscribe(subid string) {
 	s.mu.Unlock()
 }
 
-func (s *NotificationTestService) SomeSubscription(ctx context.Context, n, val int) (Subscription, error) {
+func (s *NotificationTestService) SomeSubscription(ctx context.Context, n, val int) (*Subscription, error) {
 	notifier, supported := NotifierFromContext(ctx)
 	if !supported {
 		return nil, ErrNotificationsUnsupported
@@ -59,17 +59,29 @@ func (s *NotificationTestService) SomeSubscription(ctx context.Context, n, val i
 	// by explicitly creating an subscription we make sure that the subscription id is send back to the client
 	// before the first subscription.Notify is called. Otherwise the events might be send before the response
 	// for the eth_subscribe method.
-	subscription, err := notifier.NewSubscription(s.Unsubscribe)
-	if err != nil {
-		return nil, err
-	}
+	subscription := notifier.CreateSubscription()
 
 	go func() {
+		// test expects n events, if we begin sending event immediatly some events
+		// will probably be dropped since the subscription ID might not be send to
+		// the client.
+		time.Sleep(5 * time.Second)
 		for i := 0; i < n; i++ {
-			if err := subscription.Notify(val + i); err != nil {
+			if err := notifier.Notify(subscription.ID, val+i); err != nil {
 				return
 			}
 		}
+
+		select {
+		case <-notifier.Closed():
+			s.mu.Lock()
+			s.unsubscribed = true
+			s.mu.Unlock()
+		case <-subscription.Err():
+			s.mu.Lock()
+			s.unsubscribed = true
+			s.mu.Unlock()
+		}
 	}()
 
 	return subscription, nil
@@ -77,7 +89,7 @@ func (s *NotificationTestService) SomeSubscription(ctx context.Context, n, val i
 
 // HangSubscription blocks on s.unblockHangSubscription before
 // sending anything.
-func (s *NotificationTestService) HangSubscription(ctx context.Context, val int) (Subscription, error) {
+func (s *NotificationTestService) HangSubscription(ctx context.Context, val int) (*Subscription, error) {
 	notifier, supported := NotifierFromContext(ctx)
 	if !supported {
 		return nil, ErrNotificationsUnsupported
@@ -85,12 +97,10 @@ func (s *NotificationTestService) HangSubscription(ctx context.Context, val int)
 
 	s.gotHangSubscriptionReq <- struct{}{}
 	<-s.unblockHangSubscription
-	subscription, err := notifier.NewSubscription(s.Unsubscribe)
-	if err != nil {
-		return nil, err
-	}
+	subscription := notifier.CreateSubscription()
+
 	go func() {
-		subscription.Notify(val)
+		notifier.Notify(subscription.ID, val)
 	}()
 	return subscription, nil
 }

rpc/types.go

@@ -269,6 +269,6 @@ func (bn *BlockNumber) UnmarshalJSON(data []byte) error {
 	return fmt.Errorf("blocknumber not in range [%d, %d]", earliestBlockNumber, maxBlockNumber)
 }
 
-func (bn *BlockNumber) Int64() int64 {
-	return (int64)(*bn)
+func (bn BlockNumber) Int64() int64 {
+	return (int64)(bn)
 }

rpc/utils.go

@@ -17,17 +17,26 @@
 package rpc
 
 import (
-	"crypto/rand"
+	"bufio"
+	crand "crypto/rand"
+	"encoding/binary"
 	"encoding/hex"
-	"errors"
 	"math/big"
+	"math/rand"
 	"reflect"
+	"sync"
+	"time"
 	"unicode"
 	"unicode/utf8"
 
 	"golang.org/x/net/context"
 )
 
+var (
+	subscriptionIDGenMu sync.Mutex
+	subscriptionIDGen   = idGenerator()
+)
+
 // Is this an exported - upper case - name?
 func isExported(name string) bool {
 	rune, _ := utf8.DecodeRuneInString(name)
@@ -218,11 +227,28 @@ METHODS:
 	return callbacks, subscriptions
 }
 
-func newSubscriptionID() (string, error) {
-	var subid [16]byte
-	n, _ := rand.Read(subid[:])
-	if n != 16 {
-		return "", errors.New("Unable to generate subscription id")
+// idGenerator helper utility that generates a (pseudo) random sequence of
+// bytes that are used to generate identifiers.
+func idGenerator() *rand.Rand {
+	if seed, err := binary.ReadVarint(bufio.NewReader(crand.Reader)); err == nil {
+		return rand.New(rand.NewSource(seed))
+	}
+	return rand.New(rand.NewSource(int64(time.Now().Nanosecond())))
+}
+
+// NewID generates a identifier that can be used as an identifier in the RPC interface.
+// e.g. filter and subscription identifier.
+func NewID() ID {
+	subscriptionIDGenMu.Lock()
+	defer subscriptionIDGenMu.Unlock()
+
+	id := make([]byte, 16)
+	for i := 0; i < len(id); i += 7 {
+		val := subscriptionIDGen.Int63()
+		for j := 0; i+j < len(id) && j < 7; j++ {
+			id[i+j] = byte(val)
+			val >>= 8
+		}
 	}
-	return "0x" + hex.EncodeToString(subid[:]), nil
+	return ID("0x" + hex.EncodeToString(id))
 }