pipeline

Documentation

Index

• type DeltaGetter
• type ForkHandler
  • func NewForkHandler() *ForkHandler
• type Option
  • func WithPostBlockHook(f substreams.BlockHook) Option
  • func WithPostJobHook(f substreams.PostJobHook) Option
  • func WithPreBlockHook(f substreams.BlockHook) Option
  • func WithSyncBlockRangeRestriction(maxRangeSize uint64) Option
• type Pipeline
  • func New(ctx context.Context, graph *manifest.ModuleGraph, blockType string, ...) *Pipeline
  • func (p *Pipeline) Init(ctx context.Context) (err error)
  • func (p *Pipeline) Launch(ctx context.Context, stream Streamable, streamSrv Trailable) error
  • func (p *Pipeline) ProcessBlock(block *bstream.Block, obj interface{}) (err error)
• type PipelineOptioner
• type StoreBoundary
  • func NewStoreBoundary(interval uint64, requestStopBlock uint64) *StoreBoundary
  • func (r *StoreBoundary) BumpBoundary()
  • func (r *StoreBoundary) GetStoreFlushRanges(isSubrequest bool, reqStopBlockNum uint64, blockNum uint64) []uint64
  • func (r *StoreBoundary) InitBoundary(blockNum uint64)
  • func (r *StoreBoundary) OverBoundary(blockNUm uint64) bool
• type Streamable
• type Trailable
• type UndoHandler

Types

type DeltaGetter

type DeltaGetter interface {
	GetDeltas() []*pbsubstreams.StoreDelta
}

type ForkHandler

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

func NewForkHandler

func NewForkHandler() *ForkHandler

type Option

type Option func(p *Pipeline)

func WithPostBlockHook

func WithPostBlockHook(f substreams.BlockHook) Option

func WithPostJobHook

func WithPostJobHook(f substreams.PostJobHook) Option

func WithPreBlockHook

func WithPreBlockHook(f substreams.BlockHook) Option

func WithSyncBlockRangeRestriction

func WithSyncBlockRangeRestriction(maxRangeSize uint64) Option

type Pipeline

type Pipeline struct {
	StoreMap store.Map
	// contains filtered or unexported fields
}

func New

func New(
	ctx context.Context,
	graph *manifest.ModuleGraph,
	blockType string,
	wasmExtensions []wasm.WASMExtensioner,
	execOutputCache execout.CacheEngine,
	runtimeConfig config.RuntimeConfig,
	bounder *StoreBoundary,
	respFunc func(resp *pbsubstreams.Response) error,
	opts ...Option,
) *Pipeline

func (*Pipeline) Init

func (p *Pipeline) Init(ctx context.Context) (err error)

func (*Pipeline) Launch

func (p *Pipeline) Launch(ctx context.Context, stream Streamable, streamSrv Trailable) error

func (*Pipeline) ProcessBlock

func (p *Pipeline) ProcessBlock(block *bstream.Block, obj interface{}) (err error)

type PipelineOptioner

type PipelineOptioner interface {
	PipelineOptions(ctx context.Context, request *pbsubstreams.Request) []Option
}

type StoreBoundary

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

func NewStoreBoundary

func NewStoreBoundary(
	interval uint64,
	requestStopBlock uint64,
) *StoreBoundary

func (*StoreBoundary) BumpBoundary

func (r *StoreBoundary) BumpBoundary()

func (*StoreBoundary) GetStoreFlushRanges

func (r *StoreBoundary) GetStoreFlushRanges(isSubrequest bool, reqStopBlockNum uint64, blockNum uint64) []uint64

func (*StoreBoundary) InitBoundary

func (r *StoreBoundary) InitBoundary(blockNum uint64)

func (*StoreBoundary) OverBoundary

func (r *StoreBoundary) OverBoundary(blockNUm uint64) bool

type Streamable

type Streamable interface {
	Run(ctx context.Context) error
}

type Trailable

type Trailable interface {
	SetTrailer(metadata.MD)
}

type UndoHandler

type UndoHandler func(clock *pbsubstreams.Clock, moduleOutput *pbsubstreams.ModuleOutput)