blockpoller

Documentation

Index

• type BlockFetcher
• type BlockHandler
• type BlockItem
• type BlockPoller
  • func New[C any](blockFetcher BlockFetcher[C], blockHandler BlockHandler, ...) *BlockPoller[C]
  • func (p *BlockPoller[C]) Run(firstStreamableBlockNum uint64, stopBlock *uint64, blockFetchBatchSize int) error
• type FetchResponse
• type FireBlockHandler
  • func NewFireBlockHandler(blockTypeURL string) *FireBlockHandler
  • func (f *FireBlockHandler) Handle(b *pbbstream.Block) error
  • func (f *FireBlockHandler) Init()
• type HeadBlockNumberFetcher
• type Option
  • func IgnoreCursor[C any]() Option[C]
  • func WithBlockFetchRetryCount[C any](v uint64) Option[C]
  • func WithLogger[C any](logger *zap.Logger) Option[C]
  • func WithStoringState[C any](stateStorePath string) Option[C]
• type State
  • func (s State) String() string

Types

type BlockFetcher

type BlockFetcher[C any] interface {
	IsBlockAvailable(requestedSlot uint64) bool
	Fetch(ctx context.Context, client C, blkNum uint64) (b *pbbstream.Block, skipped bool, err error)
}

type BlockHandler

type BlockHandler interface {
	Init()
	Handle(blk *pbbstream.Block) error
}

type BlockItem

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

type BlockPoller

type BlockPoller[C any] struct {
	*shutter.Shutter
	// contains filtered or unexported fields
}

func New

func New[C any](
	blockFetcher BlockFetcher[C],
	blockHandler BlockHandler,
	clients *rpc.Clients[C],
	opts ...Option[C],
) *BlockPoller[C]

func (*BlockPoller[C]) Run

func (p *BlockPoller[C]) Run(firstStreamableBlockNum uint64, stopBlock *uint64, blockFetchBatchSize int) error

type FetchResponse

type FetchResponse struct {
	Block   *pbbstream.Block
	Skipped bool
}

type FireBlockHandler

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

func NewFireBlockHandler

func NewFireBlockHandler(blockTypeURL string) *FireBlockHandler

func (*FireBlockHandler) Handle

func (f *FireBlockHandler) Handle(b *pbbstream.Block) error

func (*FireBlockHandler) Init

func (f *FireBlockHandler) Init()

type HeadBlockNumberFetcher

type HeadBlockNumberFetcher[C any] interface {
	FetchHeadBlockNumber(ctx context.Context, client C) (uint64, error)
}

type Option

type Option[C any] func(*BlockPoller[C])

func IgnoreCursor

func IgnoreCursor[C any]() Option[C]

IgnoreCursor ensures the poller will ignore the cursor and start from the startBlockNum the cursor will still be saved as the poller progresses

func WithBlockFetchRetryCount

func WithBlockFetchRetryCount[C any](v uint64) Option[C]

func WithLogger

func WithLogger[C any](logger *zap.Logger) Option[C]

func WithStoringState

func WithStoringState[C any](stateStorePath string) Option[C]

type State

type State string

const (
	ContinuousSegState State = "CONTINUOUS"
	IncompleteSegState State = "INCOMPLETE"
)

func (State) String

func (s State) String() string