event

Documentation

Index

• Constants
• func GenerateSubscriptionID() (string, error)
• func PublishAll(ctx context.Context, subscribable Subscribable, subscriber string, ...) error
• func PublishWithEventID(publisher Publisher, eventID string, eventData interface{}, ...) error
• func SubscribeCallback(ctx context.Context, subscribable Subscribable, subscriber string, ...) error
• type Cache
  • func NewEventCache(publisher Publisher) *Cache
  • func (evc *Cache) Flush() error
  • func (evc *Cache) Publish(ctx context.Context, message interface{}, tags map[string]interface{}) error
• type Emitter
  • func NewEmitter(logger *logging.Logger) Emitter
• type Publisher
  • func NewNoOpPublisher() Publisher
• type Query
  • func WrapQuery(qry pubsub.Query) Query
  • func (q Query) Query() (pubsub.Query, error)
• type QueryBuilder
  • func NewQueryBuilder(queries ...string) *QueryBuilder
  • func QueryForEventID(eventID string) *QueryBuilder
  • func (qb *QueryBuilder) And(queryBuilders ...*QueryBuilder) *QueryBuilder
  • func (qb *QueryBuilder) AndContains(tag string, operand interface{}) *QueryBuilder
  • func (qb *QueryBuilder) AndEquals(tag string, operand interface{}) *QueryBuilder
  • func (qb *QueryBuilder) AndGreaterThanOrEqual(tag string, operand interface{}) *QueryBuilder
  • func (qb *QueryBuilder) AndLessThanOrEqual(tag string, operand interface{}) *QueryBuilder
  • func (qb *QueryBuilder) AndStrictlyGreaterThan(tag string, operand interface{}) *QueryBuilder
  • func (qb *QueryBuilder) AndStrictlyLessThan(tag string, operand interface{}) *QueryBuilder
  • func (qb *QueryBuilder) Query() (pubsub.Query, error)
  • func (qb *QueryBuilder) String() string
• type QueryString
  • func (qs QueryString) Query() (pubsub.Query, error)
• type Queryable
  • func MatchAllQueryable() Queryable
• type Subscribable

Constants

const (
	EventIDKey     = "EventID"
	MessageTypeKey = "MessageType"
	TxTypeKey      = "TxType"
	TxHashKey      = "TxHash"
	StackDepthKey  = "StackDepth"
)

const DefaultEventBufferCapacity = 2 << 10

Functions

func GenerateSubscriptionID

func GenerateSubscriptionID() (string, error)

func PublishAll

func PublishAll(ctx context.Context, subscribable Subscribable, subscriber string, query Queryable,
	publisher Publisher, extraTags map[string]interface{}) error

func PublishWithEventID

func PublishWithEventID(publisher Publisher, eventID string, eventData interface{},
	extraTags map[string]interface{}) error

func SubscribeCallback

func SubscribeCallback(ctx context.Context, subscribable Subscribable, subscriber string, query Queryable,
	callback func(message interface{}) bool) error

Subscribe to messages matching query and launch a goroutine to run a callback for each one. The goroutine will exit when the context is done or the subscription is removed.

Types

type Cache

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

A Cache buffers events for a Publisher.

func NewEventCache

func NewEventCache(publisher Publisher) *Cache

Create a new Cache with an EventSwitch as backend

func (*Cache) Flush

func (evc *Cache) Flush() error

Clears cached events by flushing them to Publisher

func (*Cache) Publish

func (evc *Cache) Publish(ctx context.Context, message interface{}, tags map[string]interface{}) error

Cache an event to be fired upon finality.

type Emitter

type Emitter interface {
	Subscribable
	Publisher
	process.Process
}

func NewEmitter

func NewEmitter(logger *logging.Logger) Emitter

type Publisher

type Publisher interface {
	Publish(ctx context.Context, message interface{}, tags map[string]interface{}) error
}

func NewNoOpPublisher

func NewNoOpPublisher() Publisher

NoOpPublisher

type Query

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

A pre-parsed query

func WrapQuery

func WrapQuery(qry pubsub.Query) Query

func (Query) Query

func (q Query) Query() (pubsub.Query, error)

type QueryBuilder

type QueryBuilder struct {

	// reusable buffer for building queryString
	bytes.Buffer
	// contains filtered or unexported fields
}

A fluent query builder

func NewQueryBuilder

func NewQueryBuilder(queries ...string) *QueryBuilder

Creates a new query builder with a base query that is the conjunction of all queries passed

func QueryForEventID

func QueryForEventID(eventID string) *QueryBuilder

Get a query that matches events with a specific eventID

func (*QueryBuilder) And

func (qb *QueryBuilder) And(queryBuilders ...*QueryBuilder) *QueryBuilder

Creates the conjunction of QueryBuilder and rightQuery

func (*QueryBuilder) AndContains

func (qb *QueryBuilder) AndContains(tag string, operand interface{}) *QueryBuilder

func (*QueryBuilder) AndEquals

func (qb *QueryBuilder) AndEquals(tag string, operand interface{}) *QueryBuilder

Creates the conjunction of QueryBuilder and tag = operand

func (*QueryBuilder) AndGreaterThanOrEqual

func (qb *QueryBuilder) AndGreaterThanOrEqual(tag string, operand interface{}) *QueryBuilder

func (*QueryBuilder) AndLessThanOrEqual

func (qb *QueryBuilder) AndLessThanOrEqual(tag string, operand interface{}) *QueryBuilder

func (*QueryBuilder) AndStrictlyGreaterThan

func (qb *QueryBuilder) AndStrictlyGreaterThan(tag string, operand interface{}) *QueryBuilder

func (*QueryBuilder) AndStrictlyLessThan

func (qb *QueryBuilder) AndStrictlyLessThan(tag string, operand interface{}) *QueryBuilder

func (*QueryBuilder) Query

func (qb *QueryBuilder) Query() (pubsub.Query, error)

func (*QueryBuilder) String

func (qb *QueryBuilder) String() string

type QueryString

type QueryString string

A yet-to-parsed query

func (QueryString) Query

func (qs QueryString) Query() (pubsub.Query, error)

type Queryable

type Queryable interface {
	Query() (pubsub.Query, error)
}

func MatchAllQueryable

func MatchAllQueryable() Queryable

type Subscribable

type Subscribable interface {
	// Subscribe to all events matching query, which is a valid tmlibs Query
	Subscribe(ctx context.Context, subscriber string, query Queryable, out chan<- interface{}) error
	// Unsubscribe subscriber from a specific query string
	Unsubscribe(ctx context.Context, subscriber string, query Queryable) error
	UnsubscribeAll(ctx context.Context, subscriber string) error
}