events

Documentation

Index

• Constants
• func Consume[T any](subscriber <-chan T, process func(event T)) jobs.Job
• func Filter[T proto.Message]() func(e ABCIEventWithHeight) bool
• func QueryEventByAttributes(eventType string, module string, attributes ...sdk.Attribute) func(event Event) bool
• type ABCIEventWithHeight
  • func (e *ABCIEventWithHeight) Marshal() (dAtA []byte, err error)
  • func (e *ABCIEventWithHeight) Unmarshal(dAta []byte) error
• type AttributeValueSet
  • func NewAttributeValueSet(key string, values ...string) AttributeValueSet
  • func (s AttributeValueSet) Match(e Event) bool
• type BlockClient
• type BlockNotifier
• type BlockResultClient
• type BlockSource
  • func NewBlockSource(client BlockResultClient, notifier BlockNotifier, options ...DialOption) BlockSource
• type Bus
  • func NewEventBus(source BlockSource, bus pubsub.Bus[ABCIEventWithHeight]) *Bus
  • func (b *Bus) Done() <-chan struct{}
  • func (b *Bus) FetchEvents(ctx context.Context) <-chan error
  • func (b *Bus) Subscribe(predicate func(ABCIEventWithHeight) bool) <-chan ABCIEventWithHeight
• type DialOption
  • func BackOff(backOff time.Duration) DialOption
  • func KeepAlive(interval time.Duration) DialOption
  • func Retries(retries int) DialOption
  • func Timeout(timeout time.Duration) DialOption
• type Event
  • func Map(event ABCIEventWithHeight) Event
• type Notifier
  • func NewBlockNotifier(client BlockClient, options ...DialOption) *Notifier
  • func (b *Notifier) BlockHeights(ctx context.Context) (<-chan int64, <-chan error)
  • func (b *Notifier) Done() <-chan struct{}
  • func (b *Notifier) StartingAt(block int64) *Notifier
• type SubscriptionClient
• type SyncInfoClient

Constants

const WebsocketQueueSize = 32768

WebsocketQueueSize is set to the maximum length to prevent Tendermint from closing the connection due to congestion

Functions

func Consume

func Consume[T any](subscriber <-chan T, process func(event T)) jobs.Job

Consume processes all events from the given subscriber with the given function. Do not consume the same subscriber multiple times.

func Filter

func Filter[T proto.Message]() func(e ABCIEventWithHeight) bool

Filter returns true if an event is of the given generic type, false otherwise

func QueryEventByAttributes

func QueryEventByAttributes(eventType string, module string, attributes ...sdk.Attribute) func(event Event) bool

QueryEventByAttributes creates a Query for an event with the given attributes

Types

type ABCIEventWithHeight

type ABCIEventWithHeight struct {
	Height int64
	abci.Event
}

ABCIEventWithHeight adds a height field to abci.Event

func (*ABCIEventWithHeight) Marshal

func (e *ABCIEventWithHeight) Marshal() (dAtA []byte, err error)

Marshal extends the Marshal function of abci.Event to ABCIEventWithHeight

func (*ABCIEventWithHeight) Unmarshal

func (e *ABCIEventWithHeight) Unmarshal(dAta []byte) error

Unmarshal extends the Unmarshal function of abci.Event to ABCIEventWithHeight

type AttributeValueSet

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

AttributeValueSet represents a set of possible values for an Attribute key

func NewAttributeValueSet

func NewAttributeValueSet(key string, values ...string) AttributeValueSet

NewAttributeValueSet creates a set of possible values for an Attribute key from a list of strings

func (AttributeValueSet) Match

func (s AttributeValueSet) Match(e Event) bool

Match checks whether the passed event contains an attribute whose value is contained by the set

type BlockClient

type BlockClient interface {
	SyncInfoClient
	SubscriptionClient
}

BlockClient is both SyncInfoClient and SubscriptionClient

type BlockNotifier

type BlockNotifier interface {
	BlockHeights(ctx context.Context) (<-chan int64, <-chan error)
	Done() <-chan struct{}
}

BlockNotifier notifies the caller of new blocks

type BlockResultClient

type BlockResultClient interface {
	BlockResults(ctx context.Context, height *int64) (*coretypes.ResultBlockResults, error)
}

BlockResultClient can query for the block results of a specific block

type BlockSource

type BlockSource interface {
	BlockResults(ctx context.Context) (<-chan *coretypes.ResultBlockResults, <-chan error)
	Done() <-chan struct{}
}

BlockSource returns all block results sequentially

func NewBlockSource

func NewBlockSource(client BlockResultClient, notifier BlockNotifier, options ...DialOption) BlockSource

NewBlockSource returns a new BlockSource instance. It accepts dial options that configure the request management behaviour.

type Bus

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

Bus represents an object that receives blocks from a tendermint server and manages queries for events in those blocks

func NewEventBus

func NewEventBus(source BlockSource, bus pubsub.Bus[ABCIEventWithHeight]) *Bus

NewEventBus returns a new event bus instance

func (*Bus) Done

func (b *Bus) Done() <-chan struct{}

Done returns a channel that gets closed when the Bus is done cleaning up

func (*Bus) FetchEvents

func (b *Bus) FetchEvents(ctx context.Context) <-chan error

FetchEvents asynchronously queries the blockchain for new blocks and publishes all txs events in those blocks to the event manager's subscribers. Any occurring errors are pushed into the returned error channel.

func (*Bus) Subscribe

func (b *Bus) Subscribe(predicate func(ABCIEventWithHeight) bool) <-chan ABCIEventWithHeight

Subscribe returns an event subscription based on the given query

type DialOption

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

DialOption for Tendermint connections

func BackOff

func BackOff(backOff time.Duration) DialOption

BackOff sets the time to wait until retrying a failed call to Tendermint

func KeepAlive

func KeepAlive(interval time.Duration) DialOption

KeepAlive sets the time after which contact to Tendermint is reestablished if no there is no communication

func Retries

func Retries(retries int) DialOption

Retries sets the number of times a Tendermint call is retried

func Timeout

func Timeout(timeout time.Duration) DialOption

Timeout sets the time after which the call to Tendermint is cancelled

type Event

type Event struct {
	Type       string
	Attributes map[string]string
	Height     int64
}

Event stores type, module, action and attributes list of sdk Deprecated

func Map

func Map(event ABCIEventWithHeight) Event

Map transforms the ABCIEventWithHeight into an Event Deprecated

type Notifier

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

Notifier can notify a consumer about new blocks

func NewBlockNotifier

func NewBlockNotifier(client BlockClient, options ...DialOption) *Notifier

NewBlockNotifier returns a new BlockNotifier instance. It accepts dial options that configure the request management behaviour.

func (*Notifier) BlockHeights

func (b *Notifier) BlockHeights(ctx context.Context) (<-chan int64, <-chan error)

BlockHeights returns a channel with the block heights from the beginning of the chain to all newly discovered blocks. Optionally, starts at the given start block.

func (*Notifier) Done

func (b *Notifier) Done() <-chan struct{}

Done returns a channel that is closed when the Notifier has completed cleanup

func (*Notifier) StartingAt

func (b *Notifier) StartingAt(block int64) *Notifier

StartingAt sets the start block from which to receive notifications

type SubscriptionClient

type SubscriptionClient interface {
	Subscribe(ctx context.Context, subscriber, query string, outCapacity ...int) (out <-chan coretypes.ResultEvent, err error)
	Unsubscribe(ctx context.Context, subscriber, query string) error
}

SubscriptionClient subscribes to and unsubscribes from Tendermint events

type SyncInfoClient

type SyncInfoClient interface {
	LatestSyncInfo(ctx context.Context) (*coretypes.SyncInfo, error)
}

SyncInfoClient can query the node's sync info