poller

package

v0.0.12 Latest Latest Go to latest Published: Apr 19, 2023 License: MIT Imports: 10 Imported by: 0

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

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

Variables ¶

This section is empty.

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"`
	CursorKey   string               `env:"CURSOR_KEY" envDefault:""`
}

type Driver ¶

type Driver interface {
	Blockchain() string
	GetChainTipNumber(ctx context.Context) (uint64, error)
	IsValidBlock(ctx context.Context, index uint64) error
	FetchSequence(index uint64) map[string]pool.Runner
	Accumulate(res interface{}) pool.Runner
	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()

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL