From 271ebfa8d9e9dda6683d344c490ba58d09853669 Mon Sep 17 00:00:00 2001 From: Bas van Kervel Date: Thu, 11 Feb 2016 10:50:51 +0100 Subject: [PATCH] eth: add subscription for logs --- eth/filters/api.go | 106 +++++++++++++++++++++++++++++++---- eth/filters/filter_system.go | 2 +- rpc/server.go | 14 ++--- rpc/types.go | 12 ++-- 4 files changed, 109 insertions(+), 25 deletions(-) diff --git a/eth/filters/api.go b/eth/filters/api.go index e6a1ce3abf..ce76edb1d7 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -33,6 +33,8 @@ import ( "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" ) var ( @@ -201,8 +203,9 @@ func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) { return externalId, nil } -// newLogFilter creates a new log filter. -func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash) (int, error) { +// newLogFilter creates a new log filter. If emit is not nil logs are published +// over the channel, otherwise logs are queued and can be obtained on the next poll. +func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash, emit chan<- *event.Event) (int, error) { s.logMu.Lock() defer s.logMu.Unlock() @@ -219,11 +222,14 @@ func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []commo filter.SetAddresses(addresses) filter.SetTopics(topics) filter.LogCallback = func(log *vm.Log, removed bool) { - s.logMu.Lock() - defer s.logMu.Unlock() - - if queue := s.logQueue[id]; queue != nil { - queue.add(vmlog{log, removed}) + if emit == nil { // queue logs for next poll + s.logMu.Lock() + defer s.logMu.Unlock() + if queue := s.logQueue[id]; queue != nil { + queue.add(vmlog{log, removed}) + } + } else { + emit <- &event.Event{Time: time.Now(), Data: vmlog{log, removed}} } } @@ -364,10 +370,11 @@ func (s *PublicFilterAPI) NewFilter(args NewFilterArgs) (string, error) { var id int if len(args.Addresses) > 0 { - id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics) + id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), args.Addresses, args.Topics, nil) } else { - id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics) + id, err = s.newLogFilter(args.FromBlock.Int64(), args.ToBlock.Int64(), nil, args.Topics, nil) } + if err != nil { return "", err } @@ -379,6 +386,64 @@ func (s *PublicFilterAPI) NewFilter(args NewFilterArgs) (string, error) { return externalId, nil } +// logsSubscription implements the event.Event interface and wraps a log subscription. +type logsSubscription struct { + api *PublicFilterAPI + filterId string + logs chan *event.Event +} + +// Chan returns the channel where logs are published on for this subscription. +func (sub *logsSubscription) Chan() <-chan *event.Event { + return sub.logs +} + +// Unsubscribe will uninstall the filter and closes the emit channel +func (sub *logsSubscription) Unsubscribe() { + if sub.api.UninstallFilter(sub.filterId) { + close(sub.logs) + } +} + +// Logs creates a subscription and emits logs that match the given criteria. +func (s *PublicFilterAPI) Logs(args NewFilterArgs) (rpc.Subscription, error) { + externalId, err := newFilterId() + if err != nil { + return rpc.Subscription{}, err + } + + logs := make(chan *event.Event) + + // don't use earliest and latest since these are not applicable for subscriptions + var id int + if len(args.Addresses) > 0 { + id, err = s.newLogFilter(-1, -1, args.Addresses, args.Topics, logs) + } else { + id, err = s.newLogFilter(-1, -1, nil, args.Topics, logs) + } + + if err != nil { + return rpc.Subscription{}, err + } + + s.filterMapMu.Lock() + s.filterMapping[externalId] = id + s.filterMapMu.Unlock() + + sub := &logsSubscription{s, externalId, logs} + + format := func(rawLog interface{}) interface{} { + if log, ok := rawLog.(vmlog); ok { + rpcLogs := toRPCLogs(vm.Logs{log.Log}, log.Removed) + return &rpcLogs[0] + } + glog.V(logger.Warn).Infof("unexpected log type %T\n", rawLog) + return nil + } + + return rpc.NewSubscriptionWithOutputFormat(sub, format), nil +} + // GetLogs returns the logs matching the given argument. func (s *PublicFilterAPI) GetLogs(args NewFilterArgs) []vmlog { filter := New(s.chainDb) @@ -514,6 +579,22 @@ type vmlog struct { Removed bool `json:"removed"` } +func (l *vmlog) MarshalJSON() ([]byte, error) { + fields := map[string]interface{}{ + "address": l.Address, + "data": fmt.Sprintf("%#x", l.Data), + "blockNumber": fmt.Sprintf("%#x", l.BlockNumber), + "logIndex": fmt.Sprintf("%#x", l.Index), + "blockHash": l.BlockHash, + "transactionHash": l.TxHash, + "transactionIndex": fmt.Sprintf("%#x", l.TxIndex), + "topics": l.Topics, + "removed": l.Removed, + } + + return json.Marshal(fields) +} + type logQueue struct { mu sync.Mutex @@ -564,9 +645,10 @@ func (l *hashQueue) get() []common.Hash { return tmp } -// newFilterId generates a new random filter identifier that can be exposed to the outer world. By publishing random -// identifiers it is not feasible for DApp's to guess filter id's for other DApp's and uninstall or poll for them -// causing the affected DApp to miss data. +// newFilterId generates a new random filter identifier that can be exposed to +// the world. By publishing random identifiers it is not feasible for DApp's +// to guess filter id's for other DApp's and uninstall or poll for them causing +// the targeted DApp to miss data. func newFilterId() (string, error) { var subid [16]byte n, _ := rand.Read(subid[:]) diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index b61a493b62..5a894b55d6 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -164,7 +164,7 @@ func (fs *FilterSystem) filterLoop() { fs.filterMu.RLock() for _, filter := range fs.logFilters { if filter.LogCallback != nil && !filter.created.After(event.Time) { - for _, removedLog := range ev.Logs { + for _, removedLog := range filter.FilterLogs(ev.Logs) { filter.LogCallback(removedLog, true) } } diff --git a/rpc/server.go b/rpc/server.go index f42ee2d37a..ef1ddcca67 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -193,14 +193,14 @@ func (s *Server) Stop() { } } -// sendNotification will create a notification from the given event by serializing member fields of the event. -// It will then send the notification to the client, when it fails the codec is closed. When the event has multiple -// fields an array of values is returned. +// sendNotification will send a notification to the client when the given event +// is not nil. It will close the codec when the subscription could not be send. func sendNotification(codec ServerCodec, subid string, event interface{}) { - notification := codec.CreateNotification(subid, event) - - if err := codec.Write(notification); err != nil { - codec.Close() + if event != nil { + notification := codec.CreateNotification(subid, event) + if err := codec.Write(notification); err != nil { + codec.Close() + } } } diff --git a/rpc/types.go b/rpc/types.go index f268d84db2..c1cd7d98ca 100644 --- a/rpc/types.go +++ b/rpc/types.go @@ -136,9 +136,9 @@ func defaultSubscriptionOutputFormatter(data interface{}) interface{} { // Subscription is used by the server to send notifications to the client type Subscription struct { - sub event.Subscription - match SubscriptionMatcher - format SubscriptionOutputFormat + sub event.Subscription // event generator + match SubscriptionMatcher // allows for event filtering + format SubscriptionOutputFormat // allows for event output formatting (e.g. add custom fields) } // NewSubscription create a new RPC subscription @@ -146,13 +146,15 @@ func NewSubscription(sub event.Subscription) Subscription { return Subscription{sub, nil, defaultSubscriptionOutputFormatter} } -// NewSubscriptionWithOutputFormat create a new RPC subscription which a custom notification output format +// NewSubscriptionWithOutputFormat creates a new RPC subscription which a custom notification output format. +// If formatter returns nil the event is discarded. func NewSubscriptionWithOutputFormat(sub event.Subscription, formatter SubscriptionOutputFormat) Subscription { return Subscription{sub, nil, formatter} } // NewSubscriptionFiltered will create a new subscription. For each raised event the given matcher is -// called. If it returns true the event is send as notification to the client, otherwise it is ignored. +// called. If it returns true the event is send to the client, otherwise it is discarded. This can be +// used in situations where custom filtering is necessary. func NewSubscriptionFiltered(sub event.Subscription, match SubscriptionMatcher) Subscription { return Subscription{sub, match, defaultSubscriptionOutputFormatter} }