ingestion

Documentation

Index

• Variables
• 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) *RPCSubscriber
  • func (r *RPCSubscriber) Subscribe(ctx context.Context, height uint64) (<-chan flow.BlockEvents, <-chan error, error)

Variables

var ErrDisconnected = errors.New("disconnected")

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,
	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 event ingestion engine. Load the latest height that was stored and provide it to the event subscribers as a starting point. Consume the events provided by the event subscriber. Each event is then processed by the event processing methods.

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) *RPCSubscriber

func (*RPCSubscriber) Subscribe

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