processors

Documentation

Index

• func DecodeExecutingMessageLog(l *ethTypes.Log) (*types.ExecutingMessage, error)
• func NewEthClient(ctx context.Context, logger log.Logger, m caching.Metrics, rpc string, ...) (*sources.L1Client, error)
• type BlockProcessorFn
  • func (fn BlockProcessorFn) ProcessBlock(ctx context.Context, block eth.BlockRef) error
• type ChainProcessor
  • func NewChainProcessor(log log.Logger, chain types.ChainID, processor LogProcessor, ...) *ChainProcessor
  • func (s *ChainProcessor) Close()
  • func (s *ChainProcessor) OnNewHead(head eth.BlockRef) error
  • func (s *ChainProcessor) ProcessToHead()
  • func (s *ChainProcessor) SetSource(cl Source)
  • func (s *ChainProcessor) StartBackground()
• type ChainsDBClientForLogProcessor
• type DatabaseRewinder
• type EventDecoderFn
• type LogProcessor
  • func NewLogProcessor(chain types.ChainID, logStore LogStorage) LogProcessor
• type LogStorage
• type Source

Functions

func DecodeExecutingMessageLog

func DecodeExecutingMessageLog(l *ethTypes.Log) (*types.ExecutingMessage, error)

func NewEthClient

func NewEthClient(ctx context.Context, logger log.Logger, m caching.Metrics, rpc string, rpcClient client.RPC,
	pollRate time.Duration, trustRPC bool, kind sources.RPCProviderKind) (*sources.L1Client, error)

NewEthClient creates an Eth RPC client for event-log fetching.

Types

type BlockProcessorFn

type BlockProcessorFn func(ctx context.Context, block eth.BlockRef) error

func (BlockProcessorFn) ProcessBlock

func (fn BlockProcessorFn) ProcessBlock(ctx context.Context, block eth.BlockRef) error

type ChainProcessor

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

ChainProcessor is a HeadProcessor that fills in any skipped blocks between head update events. It ensures that, absent reorgs, every block in the chain is processed even if some head advancements are skipped.

func NewChainProcessor

func NewChainProcessor(log log.Logger, chain types.ChainID, processor LogProcessor, rewinder DatabaseRewinder, onIndexed func()) *ChainProcessor

func (*ChainProcessor) Close

func (s *ChainProcessor) Close()

func (*ChainProcessor) OnNewHead

func (s *ChainProcessor) OnNewHead(head eth.BlockRef) error

func (*ChainProcessor) ProcessToHead

func (s *ChainProcessor) ProcessToHead()

func (*ChainProcessor) SetSource

func (s *ChainProcessor) SetSource(cl Source)

func (*ChainProcessor) StartBackground

func (s *ChainProcessor) StartBackground()

type ChainsDBClientForLogProcessor

type ChainsDBClientForLogProcessor interface {
	SealBlock(chain types.ChainID, block eth.BlockRef) error
	AddLog(chain types.ChainID, logHash common.Hash, parentBlock eth.BlockID, logIdx uint32, execMsg *types.ExecutingMessage) error
}

type DatabaseRewinder

type DatabaseRewinder interface {
	Rewind(chain types.ChainID, headBlockNum uint64) error
	LatestBlockNum(chain types.ChainID) (num uint64, ok bool)
}

type EventDecoderFn

type EventDecoderFn func(*ethTypes.Log) (*types.ExecutingMessage, error)

type LogProcessor

type LogProcessor interface {
	ProcessLogs(ctx context.Context, block eth.BlockRef, receipts gethtypes.Receipts) error
}

func NewLogProcessor

func NewLogProcessor(chain types.ChainID, logStore LogStorage) LogProcessor

type LogStorage

type LogStorage interface {
	SealBlock(chain types.ChainID, block eth.BlockRef) error
	AddLog(chain types.ChainID, logHash common.Hash, parentBlock eth.BlockID, logIdx uint32, execMsg *types.ExecutingMessage) error
}

type Source

type Source interface {
	L1BlockRefByNumber(ctx context.Context, number uint64) (eth.L1BlockRef, error)
	FetchReceipts(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, gethtypes.Receipts, error)
}