processors

Documentation

Index

• func DecodeExecutingMessageLog(l *ethTypes.Log, depSet depset.ChainIndexFromID) (*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(systemContext context.Context, log log.Logger, chain eth.ChainID, ...) *ChainProcessor
  • func (s *ChainProcessor) AttachEmitter(em event.Emitter)
  • func (s *ChainProcessor) OnEvent(ev event.Event) bool
  • func (s *ChainProcessor) SetSource(cl Source)
• type DatabaseRewinder
• type EventDecoderFn
• type LogProcessor
  • func NewLogProcessor(chain eth.ChainID, logStore LogStorage, depSet depset.ChainIndexFromID) LogProcessor
• type LogStorage
• type Source

Functions

func DecodeExecutingMessageLog

func DecodeExecutingMessageLog(l *ethTypes.Log, depSet depset.ChainIndexFromID) (*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(systemContext context.Context, log log.Logger, chain eth.ChainID, processor LogProcessor, rewinder DatabaseRewinder) *ChainProcessor

func (*ChainProcessor) AttachEmitter

func (s *ChainProcessor) AttachEmitter(em event.Emitter)

func (*ChainProcessor) OnEvent

func (s *ChainProcessor) OnEvent(ev event.Event) bool

func (*ChainProcessor) SetSource

func (s *ChainProcessor) SetSource(cl Source)

type DatabaseRewinder

type DatabaseRewinder interface {
	Rewind(chain eth.ChainID, headBlock eth.BlockID) error
	LatestBlockNum(chain eth.ChainID) (num uint64, ok bool)
	AcceptedBlock(chainID eth.ChainID, id eth.BlockID) error
}

type EventDecoderFn

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

type LogProcessor

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

func NewLogProcessor

func NewLogProcessor(chain eth.ChainID, logStore LogStorage, depSet depset.ChainIndexFromID) LogProcessor

type LogStorage

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

type Source

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