stream

Documentation

Index

• Constants
• Variables
• type ErrInvalidArg
  • func NewErrInvalidArg(m string, args ...interface{}) *ErrInvalidArg
  • func (e *ErrInvalidArg) Error() string
• type Option
  • func WithBlockIndexProvider(p bstream.BlockIndexProvider) Option
  • func WithCursor(cursor *bstream.Cursor) Option
  • func WithCustomStepTypeFilter(step bstream.StepType) Option
  • func WithFileSourceHandlerMiddleware(mw func(source bstream.Handler) bstream.Handler) Option
  • func WithFinalBlocksOnly() Option
  • func WithLiveSourceHandlerMiddleware(mw func(source bstream.Handler) bstream.Handler) Option
  • func WithLogger(logger *zap.Logger) Option
  • func WithPreprocessFunc(pp bstream.PreprocessFunc, threads int) Option
  • func WithPreprocessFuncDefaultThreadNumber(pp bstream.PreprocessFunc) Option
  • func WithStopBlock(stopBlockNum uint64) Option
  • func WithTargetCursor(cursor *bstream.Cursor) Option
• type Stream
  • func New(forkedBlocksStore dstore.Store, mergedBlocksStore dstore.Store, ...) *Stream
  • func (s *Stream) Run(ctx context.Context) error

Constants

const DefaultPreprocessFuncThreadNumber = 4

Variables

var ErrStopBlockReached = errors.New("stop block reached")

Types

type ErrInvalidArg

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

func NewErrInvalidArg

func NewErrInvalidArg(m string, args ...interface{}) *ErrInvalidArg

func (*ErrInvalidArg) Error

func (e *ErrInvalidArg) Error() string

type Option

type Option = func(s *Stream)

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 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

type Stream

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

func New

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

func (*Stream) Run

func (s *Stream) Run(ctx context.Context) error