Documentation
¶
Index ¶
- Variables
- func NewStage(name StageName, runner StageRunner) *stage
- type Logger
- type Options
- type Payload
- type PayloadFactory
- type Pipeline
- func (p *Pipeline) AddStageAfter(existingStageName StageName, name StageName, stageRunner StageRunner)
- func (p *Pipeline) AddStageBefore(existingStageName StageName, name StageName, stageRunner StageRunner)
- func (p *Pipeline) Run(ctx context.Context, height int64, options *Options) (Payload, error)
- func (p *Pipeline) SetLogger(l Logger)
- func (p *Pipeline) SetOptions(o *Options)
- func (p *Pipeline) SetStage(stageName StageName, stageRunner StageRunner)
- func (p *Pipeline) Start(ctx context.Context, source Source, sink Sink, options *Options) error
- type Sink
- type Source
- type Stage
- type StageName
- type StageRunner
- type StageRunnerFunc
- type Stat
- type StatsRecorder
- type Task
- type TaskName
- type TaskValidator
Constants ¶
This section is empty.
Variables ¶
var (
ErrMissingStages = errors.New("provide stages to run concurrently")
)
Functions ¶
func NewStage ¶
func NewStage(name StageName, runner StageRunner) *stage
Types ¶
type Logger ¶
type Logger interface { // Info logs info message Info(string) // Debug logs debug message Debug(string) }
Logger is implemented by types that want to hook up to logging mechanism in engine
type Payload ¶
type Payload interface { // MarkAsProcessed is invoked by the pipeline when the payloadMock // reaches the end of execution for current height MarkAsProcessed() }
Payload is implemented by values that can be sent through a pipeline.
type PayloadFactory ¶
PayloadFactory is implemented by objects which know how to create payloadMock for every height
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline implements a modular, multi-stage pipeline
func New ¶
func New(payloadFactor PayloadFactory) *Pipeline
func (*Pipeline) AddStageAfter ¶
func (p *Pipeline) AddStageAfter(existingStageName StageName, name StageName, stageRunner StageRunner)
AddStageBefore adds custom stage after existing stage
func (*Pipeline) AddStageBefore ¶
func (p *Pipeline) AddStageBefore(existingStageName StageName, name StageName, stageRunner StageRunner)
AddStageBefore adds custom stage before existing stage
func (*Pipeline) SetOptions ¶
SetOptions sets pipeline options
func (*Pipeline) SetStage ¶ added in v0.1.5
func (p *Pipeline) SetStage(stageName StageName, stageRunner StageRunner)
SetStage sets up stage runner for given stage
type Source ¶
type Source interface { // Next gets next height Next(context.Context, Payload) bool // Current returns current height Current() int64 // Err return error if any Err() error }
Source is executed before processing of individual heights. It is responsible for getting start and end height.
type StageName ¶
type StageName string
const ( // Context key used by StatsRecorder CtxStats = "stats" // Setup stage (Chore): performs setup tasks StageSetup StageName = "stage_setup" // Fetcher stage (Syncing): fetches data for indexing StageFetcher StageName = "stage_fetcher" // Parser stage (Syncing): parses and normalizes fetched data to a single structure StageParser StageName = "stage_parser" // Validator stage (Syncing): validates parsed data StageValidator StageName = "stage_validator" // Syncer stage (Syncing): saves data to datastore StageSyncer StageName = "stage_syncer" // Sequencer stage (Indexing): Creates sequences from synced data (syncable) StageSequencer StageName = "stage_sequencer" // Aggregator stage (Indexing): Creates aggregates from synced data (syncable) StageAggregator StageName = "stage_aggregator" // StagePersistor stage (Indexing): Persists data to datastore StagePersistor StageName = "stage_persistor" // Cleanup stage (Chore): Cleans up after execution StageCleanup StageName = "stage_cleanup" )
type StageRunner ¶
type StageRunner interface { // Run StageRunner Run(context.Context, Payload, TaskValidator) error }
StageRunner is implemented by types that know how to run tasks
func RetryingStageRunner ¶
func RetryingStageRunner(sr StageRunner, isTransient func(error) bool, maxRetries int) StageRunner
RetryingStageRunner implement retry mechanism for StageRunner
type StageRunnerFunc ¶
type StageRunnerFunc func(context.Context, Payload, TaskValidator) error
StageRunnerFunc is an adapter to allow the use of plain functions as StageRunner
func (StageRunnerFunc) Run ¶
func (srf StageRunnerFunc) Run(ctx context.Context, p Payload, f TaskValidator) error
Run calls f(ctx, p, f).
type Stat ¶
func (*Stat) SetCompleted ¶
SetComplete completes stat duration
type StatsRecorder ¶
type StatsRecorder struct {
Stat
}
func NewStatsRecorder ¶
func NewStatsRecorder() *StatsRecorder
StateRecorder is responsible for recording statistics during pipeline execution TODO: Add stats for every stage and every task
type Task ¶
type Task interface { //Run Task Run(context.Context, Payload) error // GetName gets name of task GetName() string }
Task is implemented by types that want to be executed inside of a stage
type TaskValidator ¶
TaskValidator is a type for validating task by provided task name