pipeline

Documentation

Index

• func BuildRequestDetails(request *pbsubstreams.Request, isSubRequest bool, ...) (req *reqctx.RequestDetails, err error)
• func NewStoreBoundary(interval uint64, requestStartBlockNum uint64, requestStopBlock uint64) *storeBoundary
• 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
• type Pipeline
  • func New(ctx context.Context, outputGraph *outputmodules.Graph, stores *Stores, ...) *Pipeline
  • func (p *Pipeline) GetStoreMap() store.Map
  • func (p *Pipeline) InitStoresAndBackprocess(ctx context.Context) (err error)
  • func (p *Pipeline) InitWASM(ctx context.Context) (err error)
  • func (p *Pipeline) OnStreamTerminated(ctx context.Context, streamSrv Trailable, err error) error
  • func (p *Pipeline) ProcessBlock(block *bstream.Block, obj interface{}) (err error)
• type PipelineOptioner
• type Stores
  • func NewStores(storeConfigs store.ConfigMap, ...) *Stores
  • func (s *Stores) SetStoreMap(storeMap store.Map)
• type Trailable
• type UndoHandler

Functions

func BuildRequestDetails

func BuildRequestDetails(request *pbsubstreams.Request, isSubRequest bool, isOutputModule reqctx.IsOutputModuleFunc, getRecentFinalBlock getBlockFunc, getHeadBlock getBlockFunc) (req *reqctx.RequestDetails, err error)

func NewStoreBoundary

func NewStoreBoundary(
	interval uint64,
	requestStartBlockNum uint64,
	requestStopBlock uint64,
) *storeBoundary

Types

type ForkHandler

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

TODO(abourget): The scope of this object and the Engine

are not pretty similar, to keep track of certain pieces
of info that are reversible, and handle the back and forth
between undos and redos.
Perhaps what we could have here, is have those undo handlers
live on the Pipeline (where it makes sense)
and have some nested structs handle

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

type Pipeline

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

func New

func New(
	ctx context.Context,
	outputGraph *outputmodules.Graph,
	stores *Stores,
	execoutStorage *execout.Configs,
	wasmRuntime *wasm.Runtime,
	execOutputCache *cache.Engine,
	runtimeConfig config.RuntimeConfig,
	respFunc func(resp *pbsubstreams.Response) error,
	opts ...Option,
) *Pipeline

func (*Pipeline) GetStoreMap

func (p *Pipeline) GetStoreMap() store.Map

func (*Pipeline) InitStoresAndBackprocess

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

func (*Pipeline) InitWASM

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

func (*Pipeline) OnStreamTerminated

func (p *Pipeline) OnStreamTerminated(ctx context.Context, streamSrv Trailable, err error) error

OnStreamTerminated performs flush of store and setting trailers when the stream terminated gracefully from our point of view. If the stream terminated gracefully, we return `nil` otherwise, the original is returned.

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 Stores

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

func NewStores

func NewStores(storeConfigs store.ConfigMap, storeSnapshotSaveInterval, requestStartBlockNum, stopBlockNum uint64, isSubRequest bool) *Stores

func (*Stores) SetStoreMap

func (s *Stores) SetStoreMap(storeMap store.Map)

type Trailable

type Trailable interface {
	SetTrailer(metadata.MD)
}

type UndoHandler

type UndoHandler func(clock *pbsubstreams.Clock, moduleOutputs []*pbsubstreams.ModuleOutput)