From 4812a4f49e78220f617737cda909c54075376cc5 Mon Sep 17 00:00:00 2001 From: Max Wolter Date: Mon, 15 Aug 2016 20:47:52 +0200 Subject: [PATCH] Refactored the subscription with the new unpacking --- accounts/abi/bind/base.go | 173 +++++++++++++++++++------------------- 1 file changed, 85 insertions(+), 88 deletions(-) diff --git a/accounts/abi/bind/base.go b/accounts/abi/bind/base.go index 3d33283288..1342f64a5d 100644 --- a/accounts/abi/bind/base.go +++ b/accounts/abi/bind/base.go @@ -60,9 +60,6 @@ type TransactOpts struct { // SubscribeOpts is the collection of options to fine tune the contract subscription // and unsubscription requests. type SubscribeOpts struct { - History bool // Whether to also retrieve events from the past - FromBlock *big.Int // Block at which we want to start retrieving past events - Context context.Context // Network context to support cancellation and timeouts (nil = no timeout) } @@ -87,17 +84,20 @@ type BoundContract struct { latestHasCode uint32 // Cached verification that the latest state contains code for this contract pendingHasCode uint32 // Cached verification that the pending state contains code for this contract - subscriptions map[chan<- vm.Log]ethereum.Subscription - errors map[chan<- vm.Log]error + subscriptions map[uint32]ethereum.Subscription + errors map[uint32]error + nextID uint32 } // NewBoundContract creates a low level contract interface through which calls // and transactions may be made through. func NewBoundContract(address common.Address, abi abi.ABI, backend ContractBackend) *BoundContract { return &BoundContract{ - address: address, - abi: abi, - backend: backend, + address: address, + abi: abi, + backend: backend, + subscriptions: make(map[uint32]ethereum.Subscription), + errors: make(map[uint32]error), } } @@ -261,6 +261,18 @@ func (c *BoundContract) transact(opts *TransactOpts, contract *common.Address, i // the given start & end blocks. func (c *BoundContract) Events(opts *EventOpts, name string, output interface{}, topics ...[]common.Hash) error { + // check that output is a pointer + ptr := reflect.ValueOf(output) + if ptr.Kind() != reflect.Ptr { + return fmt.Errorf("need pointer to slice as output parameter, have %T", output) + } + + // check that it points to a slice + slice := ptr.Elem() + if slice.Kind() != reflect.Slice { + return fmt.Errorf("need pointer to slice as output parameter, have %T", output) + } + // get the event so we can encode the name into the first topic event, ok := c.abi.Events[name] if !ok { @@ -269,18 +281,6 @@ func (c *BoundContract) Events(opts *EventOpts, name string, output interface{}, names := []common.Hash{event.Id()} topics = append([][]common.Hash{names}, topics...) - // check that output is a pointer - ptr := reflect.ValueOf(output) - if ptr.Kind() != reflect.Ptr { - return fmt.Errorf("need pointer to slice as output, have %T", output) - } - - // check that it points to a slice - slice := ptr.Elem() - if slice.Kind() != reflect.Slice { - return fmt.Errorf("need pointer to slice as output, have %T", output) - } - // create the filter query and retrieve the events through the backend filterQuery := ethereum.FilterQuery{ FromBlock: opts.FromBlock, @@ -315,102 +315,99 @@ func (c *BoundContract) Events(opts *EventOpts, name string, output interface{}, // Subscribe subscribes to the contract for the given topics. Subscription events // will be submitted to the provided channel. Upon cancelation of the subscription, -// the channel will be closed. A channel should only be used for ones subscription. -func (c *BoundContract) Subscribe(opts *SubscribeOpts, name string, channel chan<- vm.Log, topics ...[]common.Hash) error { +// the channel will be closed. A channel should only be used for one subscription. +func (c *BoundContract) Subscribe(opts *SubscribeOpts, name string, output interface{}, topics ...[]common.Hash) (uint32, error) { - // check if we already have a subscription on this channel - _, ok := c.subscriptions[channel] - if ok { - return fmt.Errorf("cannot reuse channel for multiple subscriptions") + // check that output is a pointer + channel := reflect.ValueOf(output) + if channel.Kind() != reflect.Chan { + return 0, fmt.Errorf("need channel as output parameter, have %T", output) } + message := channel.Elem() - // Build the filter query with our desired options and topics + // get the event so we can encode the name into the first topic + event, ok := c.abi.Events[name] + if !ok { + return 0, fmt.Errorf("unknown event name: %v", name) + } + names := []common.Hash{event.Id()} + topics = append([][]common.Hash{names}, topics...) + + // create the filter query and retrieve the events through the backend filterQuery := ethereum.FilterQuery{ - FromBlock: opts.FromBlock, + FromBlock: nil, ToBlock: nil, Addresses: []common.Address{c.address}, Topics: topics, } - // If we don't want to miss any real-time event logs, we need to subscribe right away - tempChannel := make(chan vm.Log) - tempSub, err := c.backend.SubscribeFilterLogs(opts.Context, filterQuery, tempChannel) + // create a channel of log messages that we can decode into events + logs := make(chan vm.Log) + subscription, err := c.backend.SubscribeFilterLogs(opts.Context, filterQuery, logs) if err != nil { - return fmt.Errorf("could not create subscription (%v)", err) + return 0, fmt.Errorf("could not subscribe (%v)", err) } + id := c.nextID + c.subscriptions[id] = subscription + c.nextID++ - // Get the desired historical data and feed it into the subscription channel - if opts.History { - var logs []vm.Log - logs, err = c.backend.FilterLogs(opts.Context, filterQuery) - if err != nil { - return fmt.Errorf("could not retrieve history of subscription events") - } - for _, log := range logs { - channel <- log - } - } + // retrieve logs and transcribe them into typed events + go func() { + Loop: + for { + select { + case err, ok = <-subscription.Err(): - // Feed the real-time events that happenend in the meantime to the subscription channel - // and create the new subscription as soon as none are left - var subscription ethereum.Subscription -Feed: - for { - select { - case log := <-tempChannel: - channel <- log - default: - subscription, err = c.backend.SubscribeFilterLogs(opts.Context, filterQuery, channel) - tempSub.Unsubscribe() - for _ = range tempSub.Err() { - } - Drain: - for { - select { - case <-tempChannel: - default: - break Drain + // error was closed, so we shut down + if !ok { + break Loop } + + // otherwise just save the error for this subscription + c.errors[id] = err + + // receive log and transcribe + case log := <-logs: + item := reflect.New(message.Type()) + c.abi.Unpack(item.Interface(), name, log.Data) + channel.Send(item) } - close(tempChannel) - break Feed } - } - // check if the subscription succeeded - if err != nil { - return fmt.Errorf("could not create subscription (%v)", err) - } - c.subscriptions[channel] = subscription - - // Read from the error channel in the background - // When we receive an error, we save it with the channel index - // When the error channel is closed, it indicates an ended subscription - // We then close the subscription channel to notify the consumer and remove - // the subscription from our map - go func(errChannel <-chan error) { - for err := range errChannel { - c.errors[channel] = err + // drain the logs channel; this might not always work correctly because we + // are not sure all logs were put in the channel already + Drain: + for { + select { + case log := <-logs: + item := reflect.New(message.Type()) + c.abi.Unpack(item.Interface(), name, log.Data) + channel.Send(item) + default: + break Drain + } } - close(channel) - delete(c.subscriptions, channel) - }(subscription.Err()) - return nil + + // close the output channel + channel.Close() + }() + + return id, nil } // Error returns the error that was encountered for the subscription on the given // channel. It will reset the error upon retrieval. -func (c *BoundContract) Error(channel chan<- vm.Log) error { - err := c.errors[channel] - delete(c.errors, channel) +func (c *BoundContract) Error(id uint32) error { + err := c.errors[id] + delete(c.errors, id) return err } // Unsubscribe cancels the subscription if one exists on the given channel. The // subscription channel will be closed once the subscription was successfully // canceled. -func (c *BoundContract) Unsubscribe(channel chan<- vm.Log) error { - subscription, ok := c.subscriptions[channel] +func (c *BoundContract) Unsubscribe(id uint32) error { + subscription, ok := c.subscriptions[id] if !ok { return fmt.Errorf("subscription doesn't exist or already closed") }