stream

func NewErrInvalidArg ¶

func NewErrInvalidArg(m string, args ...any) *ErrInvalidArg

func NewErrUnavailable ¶

func NewErrUnavailable(m string, args ...any) *ErrUnavailable

func WithBlockIndexProvider ¶

func WithBlockIndexProvider(p bstream.BlockIndexProvider) Option

func WithCursor ¶

func WithCursor(cursor *bstream.Cursor) Option

func WithCustomStepTypeFilter ¶

func WithCustomStepTypeFilter(step bstream.StepType) Option

func WithFileSourceHandlerMiddleware ¶

func WithFileSourceHandlerMiddleware(mw func(source bstream.Handler) bstream.Handler) Option

func WithFinalBlocksOnly ¶

func WithFinalBlocksOnly() Option

func WithLiveSourceHandlerMiddleware ¶

func WithLiveSourceHandlerMiddleware(mw func(source bstream.Handler) bstream.Handler) Option

func WithLogger ¶

func WithLogger(logger *zap.Logger) Option

func WithMergedBlocksBundleSize ¶

func WithMergedBlocksBundleSize(bundleSize uint64) Option

WithMergedBlocksBundleSize overrides the number of blocks per merged-blocks file for this stream. When not set, bstream.DefaultMergedBlocksBundleSize applies. The value must match the size of the files actually present in the merged-blocks store.

func WithPreprocessFunc ¶

func WithPreprocessFunc(pp bstream.PreprocessFunc, threads int) Option

func WithPreprocessFuncDefaultThreadNumber ¶

func WithPreprocessFuncDefaultThreadNumber(pp bstream.PreprocessFunc) Option

func WithStopBlock ¶

func WithStopBlock(stopBlockNum uint64) Option

func WithTargetCursor ¶

func WithTargetCursor(cursor *bstream.Cursor) Option

func WithoutPartialBlocks ¶

func WithoutPartialBlocks() Option

WithoutPartialBlocks makes the stream leave out the partial blocks of the hub, receiving each block once, complete.

func New ¶

func New(
	forkedBlocksStore dstore.Store,
	mergedBlocksStore dstore.Store,
	hub *hub.ForkableHub,
	startBlockNum int64,
	handler bstream.Handler,
	options ...Option) *Stream