ingestion

Documentation

Index

• type Engine
  • func NewEventIngestionEngine(subscriber EventSubscriber, blocks storage.BlockIndexer, ...) *Engine
  • func (e *Engine) Done() <-chan struct{}
  • func (e *Engine) Ready() <-chan struct{}
  • func (e *Engine) Run(ctx context.Context) error
  • func (e *Engine) Stop()
• type EventSubscriber
• type RPCSubscriber
  • func NewRPCSubscriber(client access.Client, chainID flowGo.ChainID) *RPCSubscriber
  • func (r *RPCSubscriber) Subscribe(ctx context.Context, height uint64) (<-chan flow.BlockEvents, <-chan error, error)

Types

type Engine

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

func NewEventIngestionEngine

func NewEventIngestionEngine(
	subscriber EventSubscriber,
	blocks storage.BlockIndexer,
	receipts storage.ReceiptIndexer,
	transactions storage.TransactionIndexer,
	accounts storage.AccountIndexer,
	blocksBroadcaster *engine.Broadcaster,
	transactionsBroadcaster *engine.Broadcaster,
	logsBroadcaster *engine.Broadcaster,
	log zerolog.Logger,
) *Engine

func (*Engine) Done

func (e *Engine) Done() <-chan struct{}

Done signals when the engine has stopped.

func (*Engine) Ready

func (e *Engine) Ready() <-chan struct{}

Ready signals when the engine has started.

func (*Engine) Run

func (e *Engine) Run(ctx context.Context) error

Run the Cadence event ingestion engine.

Cadence event ingestion engine subscribes to all new EVM related events on Flow network, currently there are two types of events: - evm.BlockExecuted: this event is emitted when a new EVM block is created (usually with any new transactions) - evm.TransactionExecuted: this event is emitted when a new EVM transaction is executed (even if failed) Each event that is received should contain at least block event, but usually also transaction events. There can be multiple transaction events for a single Cadence height, but only a single block event. Events are after indexed in-order, first block event and then all transaction events.

Expected errors: there is a disconnected error which is a recoverable error that can be expected and should be handled by restarting the engine. This can happen if the client connection to the event subscription drops. All other errors are unexpected.

func (*Engine) Stop

func (e *Engine) Stop()

Stop the engine.

type EventSubscriber

type EventSubscriber interface {
	// Subscribe to relevant events from the provided block height.
	// Returns a channel with block events and errors,
	// if subscription fails returns an error as the third value.
	Subscribe(ctx context.Context, height uint64) (<-chan flow.BlockEvents, <-chan error, error)
}

type RPCSubscriber

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

func NewRPCSubscriber

func NewRPCSubscriber(client access.Client, chainID flowGo.ChainID) *RPCSubscriber

func (*RPCSubscriber) Subscribe

func (r *RPCSubscriber) Subscribe(ctx context.Context, height uint64) (<-chan flow.BlockEvents, <-chan error, error)