watcher

Documentation

Index

• Variables
• type LogEventID
• type Watch
  • func (w *Watch) Ack() error
  • func (w *Watch) Close()
  • func (w *Watch) Next() (types.Log, error)
• type WatchClient
• type WatchDAL
• type WatchService
  • func NewWatchService(client WatchClient, dal WatchDAL, polling, maxBlockDelta uint64) *WatchService
  • func (ws *WatchService) Close()
  • func (ws *WatchService) GetBlockNumber() uint64
  • func (ws *WatchService) GetCurrentBlockNumber() *big.Int
  • func (ws *WatchService) MakeFilterQuery(addr common.Address, rawABI string, eventName string, startBlock *big.Int) (ethereum.FilterQuery, error)
  • func (ws *WatchService) NewWatch(name string, query ethereum.FilterQuery, ...) (*Watch, error)

Variables

var (
	ErrWatchServiceClosed = errors.New("Watch service closed")
	ErrWatcherTimeout     = errors.New("Watcher timeout")
)

Types

type LogEventID

type LogEventID struct {
	BlockNumber uint64 // Number of the block containing the event
	Index       int64  // Index of the event within the block
}

LogEventID tracks the position of a watch event in the event log.

type Watch

type Watch struct {
	// contains filtered or unexported fields
}

Watch provides an iterator over a stream of event logs that match an Ethereum filtering query. It updates the KVStore to persist the position in the stream of the last event log that the application has acknowledged receiving. To handle chain reorganization (ephemeral forking), watch only requests from on-chain event logs that are older than a specified number of on-chain blocks.

func (*Watch) Ack

func (w *Watch) Ack() error

The app ACKs the complete processing of the last received event log. Be lenient in one case: after the watch is closed, allow at most one more ACK to be done. This allows event processing that was completed by the application when an asynchronous Close() took place (between the Next() and the Ack() calls) to be persisted into storage instead of having it be re-done after the application is restarted.

func (*Watch) Close

func (w *Watch) Close()

Close a watch subscription.

func (*Watch) Next

func (w *Watch) Next() (types.Log, error)

Fetch the next log event. The function will block until either an event log is available, or the watcher is closed.

type WatchClient

type WatchClient interface {
	HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
	FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
}

WatchClient is an interface for the subset of functions of the Go-Ethereum client API that the watch service uses.

type WatchDAL

type WatchDAL interface {
	InsertMonitor(event string, blockNum uint64, blockIdx int64, restart bool) error
	GetMonitorBlock(event string) (uint64, int64, bool, error)
	UpdateMonitorBlock(event string, blockNum uint64, blockIdx int64) error
	UpsertMonitorBlock(event string, blockNum uint64, blockIdx int64, restart bool) error
}

WatchDAL is an interface for the watch-specific API of the KVStore data access layer.

type WatchService

type WatchService struct {
	// contains filtered or unexported fields
}

WatchService holds the active watchers and their connections to the Ethereum client and the KVStore persistence layer that provides resumability of the watcher after a restart.

func NewWatchService

func NewWatchService(client WatchClient, dal WatchDAL, polling, maxBlockDelta uint64) *WatchService

Create a watch service. polling: interval (in seconds) to periodically query eth logs maxBlockDelta: maximum number of blocks for each eth log query, 0 means unlimited

func (*WatchService) Close

func (ws *WatchService) Close()

Close the watch service.

func (*WatchService) GetBlockNumber

func (ws *WatchService) GetBlockNumber() uint64

Return the most recent on-chain block number.

func (*WatchService) GetCurrentBlockNumber

func (ws *WatchService) GetCurrentBlockNumber() *big.Int

Return the most recent on-chain block number in big.Int format.

func (*WatchService) MakeFilterQuery

func (ws *WatchService) MakeFilterQuery(
	addr common.Address, rawABI string, eventName string, startBlock *big.Int) (ethereum.FilterQuery, error)

MakeFilterQuery constructs an Ethereum FilterQuery structure from these event and contract parameters: address, raw ABI string, event name, and the optional start block number.

func (*WatchService) NewWatch

func (ws *WatchService) NewWatch(
	name string, query ethereum.FilterQuery, blkDelay, fwdDelay, checkInterval uint64, reset bool) (*Watch, error)

Create a watch for the given Ethereum log filtering query. The block delay is the number of blocks mined used as a time delay for fetching event logs, mitigating the effects of chain reorg. The block interval controls the polling frequency of fetch logs from on-chain, but measured in block numbers (as a delta). If "reset" is enabled, the watcher ignores the previously stored position in the subscription which resets the stream to its start.