eventsyncer

Documentation

Overview

Package eventsyncer implements functions for syncing registry contract events by reading them using executionclient and processing them using eventhandler.

Index

• Variables
• type EventHandler
• type EventSyncer
  • func New(nodeStorage nodestorage.Storage, executionClient ExecutionClient, ...) *EventSyncer
  • func (es *EventSyncer) Healthy(ctx context.Context) error
  • func (es *EventSyncer) SyncHistory(ctx context.Context, fromBlock uint64) (lastProcessedBlock uint64, err error)
  • func (es *EventSyncer) SyncOngoing(ctx context.Context, fromBlock uint64) error
• type ExecutionClient
• type Option
  • func WithLogger(logger *zap.Logger) Option
  • func WithMetrics(metrics metrics) Option
  • func WithStalenessThreshold(threshold time.Duration) Option

Variables

var (
	// ErrNodeNotReady is returned when node is not ready.
	ErrNodeNotReady = fmt.Errorf("node not ready")
)

Types

type EventHandler

type EventHandler interface {
	HandleBlockEventsStream(logs <-chan executionclient.BlockLogs, executeTasks bool) (uint64, error)
}

type EventSyncer

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

EventSyncer syncs registry contract events from the given ExecutionClient and passes them to the given EventHandler for processing.

func New

func New(nodeStorage nodestorage.Storage, executionClient ExecutionClient, eventHandler EventHandler, opts ...Option) *EventSyncer

func (*EventSyncer) Healthy

func (es *EventSyncer) Healthy(ctx context.Context) error

Healthy returns nil if the syncer is syncing ongoing events.

func (*EventSyncer) SyncHistory

func (es *EventSyncer) SyncHistory(ctx context.Context, fromBlock uint64) (lastProcessedBlock uint64, err error)

SyncHistory reads and processes historical events since the given fromBlock.

func (*EventSyncer) SyncOngoing

func (es *EventSyncer) SyncOngoing(ctx context.Context, fromBlock uint64) error

SyncOngoing streams and processes ongoing events as they come since the given fromBlock.

type ExecutionClient

type ExecutionClient interface {
	FetchHistoricalLogs(ctx context.Context, fromBlock uint64) (logs <-chan executionclient.BlockLogs, errors <-chan error, err error)
	StreamLogs(ctx context.Context, fromBlock uint64) <-chan executionclient.BlockLogs
}

type Option

type Option func(*EventSyncer)

Option defines EventSyncer configuration option.

func WithLogger

func WithLogger(logger *zap.Logger) Option

WithLogger enables logging.

func WithMetrics

func WithMetrics(metrics metrics) Option

WithMetrics enables reporting metrics.

func WithStalenessThreshold

func WithStalenessThreshold(threshold time.Duration) Option