contract_poller

Documentation

Index

• Constants
• func WithAccumulatePool(wp *pool.WorkerPool) opt
• func WithCache(c Cache) opt
• func WithFetchPool(wp *pool.WorkerPool) opt
• func WithLogger(logger util.Logger) opt
• func WithMetrics(metrics util.Metrics) opt
• func WithWritePool(wp *pool.WorkerPool) opt
• type Cache
• type Config
• type Driver
• type Poller
  • func New(cfg *Config, driver Driver, opts ...opt) *Poller
  • func (p *Poller) Insights() map[string]map[string]int
  • func (p *Poller) Mode() int
  • func (p *Poller) Pause()
  • func (p *Poller) Resume()
  • func (p *Poller) SetCursor(ctx context.Context, newVal uint64) error
  • func (p *Poller) Start(ctx context.Context) error
  • func (p *Poller) Stop()

Constants

const (
	//	ModeReady means the poller is ready to have its mode reassessed based on chaintip position
	ModeReady = iota
	//	ModeSleep means the poller is within reorg depth and is waiting to chaintip to progress before reassessing
	ModeSleep
	//	ModePaused means the poller has been manually paused - it will stay in this state until manually resumed
	ModePaused
	//	ModeBackfill means the poller is >= batchSize blocks from chaintip and will pull a batch of past blocks
	ModeBackfill
	//	ModeChaintip means the poller is < batchSize blocks from chaintip and will pull one block at a time
	ModeChaintip
)

Modes determine what the poller does on each iteration of its main routine's loop; these are determined by the distance of the cursor from chaintip and the success/failure state of the previous iteration

Functions

func WithAccumulatePool

func WithAccumulatePool(wp *pool.WorkerPool) opt

func WithCache

func WithCache(c Cache) opt

func WithFetchPool

func WithFetchPool(wp *pool.WorkerPool) opt

func WithLogger

func WithLogger(logger util.Logger) opt

func WithMetrics

func WithMetrics(metrics util.Metrics) opt

func WithWritePool

func WithWritePool(wp *pool.WorkerPool) opt

Types

type Cache

type Cache interface {
	GetCurrentBlockNumber(ctx context.Context, blockChainInfoKey string) (uint64, error)
	SetCurrentBlockNumber(ctx context.Context, blockChainInfoKey string, blockNumber uint64) error
}

type Config

type Config struct {
	Blockchain  constants.Blockchain `env:"BLOCKCHAIN,required"`
	BatchSize   int                  `env:"BATCH_SIZE" envDefault:"100"`
	ReorgDepth  int                  `env:"REORG_DEPTH" envDefault:"8"`
	HttpRetries int                  `env:"HTTP_RETRIES" envDefault:"10"`
	SleepTime   time.Duration        `env:"POLLER_SLEEP_TIME" envDefault:"12s"`
	Tick        time.Duration        `env:"POLLER_TICK_DURATION" envDefault:"1s"`
	AutoStart   bool                 `env:"POLLER_AUTO_START" envDefault:"false"`
}

type Driver

type Driver interface {
	Blockchain() string
	GetChainTipNumber(ctx context.Context) (uint64, error)
	// FetchSequence involves fetching trace and receipt from ethrpc node
	FetchSequence(blockNumber uint64) map[string]pool.Runner
	// Fetchers involves fetching contract's abi and metadata, using results from FetchContractAddresses
	Fetchers() map[string]pool.FeedTransformer
	// Accumulate involves combining abi, metadata to form a complete contract, building fragments
	Accumulate(res interface{}) pool.Runner
	// Writers involves writing to postgresDB
	Writers() []pool.FeedTransformer
}

type Poller

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

Poller is a chain-agnostic module for ETLing blockchain data, utilizing worker pools to optimize efficiency and speed

func New

func New(cfg *Config, driver Driver, opts ...opt) *Poller

New constructs a new poller, given a config, a chain-specific driver, and a variadic array of options

func (*Poller) Insights

func (p *Poller) Insights() map[string]map[string]int

func (*Poller) Mode

func (p *Poller) Mode() int

func (*Poller) Pause

func (p *Poller) Pause()

func (*Poller) Resume

func (p *Poller) Resume()

func (*Poller) SetCursor

func (p *Poller) SetCursor(ctx context.Context, newVal uint64) error

func (*Poller) Start

func (p *Poller) Start(ctx context.Context) error

Run executes the main program loop inside of a dedicated goroutine; the loop can be terminated from the outside via context cancellation

func (*Poller) Stop

func (p *Poller) Stop()