repo

Documentation

Index

• func Entries[inner any](reply any, err error) ([]streamEntry[inner], error)
• func StreamNotifications[inner any](reply any, err error) (map[string][]streamEntry[inner], error)
• type EventID
• type Mongo
  • func NewMongo(c *mongo.Client, l lgr.L) Mongo
  • func (m Mongo) FetchRelations(ctx context.Context, from, to string, providers []string, after string, ...) ([]graph.Relation, error)
  • func (m Mongo) SaveRelations(ctx context.Context, relations []graph.Relation) error
• type Redis
  • func NewRedis(l lgr.L, pool *redis.Pool) (r Redis)
  • func (r Redis) AcknowledgeBlocks(ctx context.Context, events []EventID) error
  • func (r Redis) AddBlocks(ctx context.Context, blocks []uint64) error
  • func (r Redis) FetchStreamEvents(ctx context.Context, consumerID string, batchSize uint) (map[EventID]uint64, error)
  • func (r Redis) FindStaleBlocks(ctx context.Context, consumerID string, timeout time.Duration, batchSize uint) (map[EventID]uint64, error)
  • func (h Redis) GetLastSeenBlock(ctx context.Context) (uint64, error)
  • func (h Redis) InitializeRedis(ctx context.Context) error
  • func (h Redis) SaveLastSeenBlock(ctx context.Context, block uint64) error

Functions

func Entries

func Entries[inner any](reply any, err error) ([]streamEntry[inner], error)

Entries is a helper that converts an array of stream entries into Entry values. Requires two values in each entry result, and an even number of field values.

func StreamNotifications

func StreamNotifications[inner any](reply any, err error) (map[string][]streamEntry[inner], error)

Types

type EventID

type EventID = string

EventID is opaque identifier used identifying and ack'ing prococessed events In Redis implementations corresponds to entry ID returned by XADD, e.g. `1518951480106-0`

type Mongo

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

func NewMongo

func NewMongo(c *mongo.Client, l lgr.L) Mongo

func (Mongo) FetchRelations

func (m Mongo) FetchRelations(ctx context.Context, from, to string, providers []string, after string, limit uint) ([]graph.Relation, error)

func (Mongo) SaveRelations

func (m Mongo) SaveRelations(ctx context.Context, relations []graph.Relation) error

type Redis

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

func NewRedis

func NewRedis(l lgr.L, pool *redis.Pool) (r Redis)

func (Redis) AcknowledgeBlocks

func (r Redis) AcknowledgeBlocks(ctx context.Context, events []EventID) error

func (Redis) AddBlocks

func (r Redis) AddBlocks(ctx context.Context, blocks []uint64) error

AddTransaction tries to add blocks to the processing stream and returns blocks that have already been scheduled (if any)

func (Redis) FetchStreamEvents

func (r Redis) FetchStreamEvents(ctx context.Context, consumerID string, batchSize uint) (map[EventID]uint64, error)

func (Redis) FindStaleBlocks

func (r Redis) FindStaleBlocks(ctx context.Context, consumerID string, timeout time.Duration, batchSize uint) (map[EventID]uint64, error)

func (Redis) GetLastSeenBlock

func (h Redis) GetLastSeenBlock(ctx context.Context) (uint64, error)

func (Redis) InitializeRedis

func (h Redis) InitializeRedis(ctx context.Context) error

func (Redis) SaveLastSeenBlock

func (h Redis) SaveLastSeenBlock(ctx context.Context, block uint64) error