event

Documentation

Index

• Constants
• func GenerateSubscriptionID() (string, error)
• func LogNKey(topic int) string
• func LogNTextKey(topic int) string
• func PublishAll(ctx context.Context, subscribable Subscribable, subscriber string, ...) error
• func QueryForEventID(eventID string) *query.Builder
• func SubscribeCallback(ctx context.Context, subscribable Subscribable, subscriber string, ...) error
• type Cache
  • func NewCache() *Cache
  • func (evc *Cache) Flush(publisher Publisher) error
  • func (evc *Cache) Publish(ctx context.Context, message interface{}, tags Tags) error
  • func (evc *Cache) Reset()
  • func (evc *Cache) Sync(publisher Publisher) error
• type CombinedTags
  • func (ct CombinedTags) Get(key string) (value string, ok bool)
  • func (ct CombinedTags) Keys() []string
  • func (ct CombinedTags) Len() (length int)
  • func (ct CombinedTags) Map() map[string]interface{}
• type Emitter
  • func NewEmitter(logger *logging.Logger) Emitter
• type Indexable
• type Publisher
  • func NewNoOpPublisher() Publisher
• type PublisherFunc
  • func (pf PublisherFunc) Publish(ctx context.Context, message interface{}, tags Tags) error
• type Subscribable
• type TagMap
  • func (ts TagMap) Get(key string) (value string, ok bool)
  • func (ts TagMap) Keys() []string
  • func (ts TagMap) Len() int
  • func (ts TagMap) Map() map[string]interface{}
• type Tags

Constants

const (
	EventTypeKey   = "EventType"
	EventIDKey     = "EventID"
	MessageTypeKey = "MessageType"
	TxTypeKey      = "TxType"
	TxHashKey      = "TxHash"
	HeightKey      = "Height"
	IndexKey       = "Index"
	NameKey        = "Name"
	PermissionKey  = "Permission"
	StackDepthKey  = "StackDepth"
	AddressKey     = "Address"
	OriginKey      = "Origin"
	CalleeKey      = "Callee"
	CallerKey      = "Caller"
	ValueKey       = "Value"
	GasKey         = "Gas"
	ExceptionKey   = "Exception"
	LogNKeyPrefix  = "Log"
)

const DefaultEventBufferCapacity = 2 << 10

const SubscribeCallbackTimeout = 2 * time.Second

Functions

func GenerateSubscriptionID

func GenerateSubscriptionID() (string, error)

func LogNKey

func LogNKey(topic int) string

func LogNTextKey

func LogNTextKey(topic int) string

func PublishAll

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

func QueryForEventID

func QueryForEventID(eventID string) *query.Builder

Get a query that matches events with a specific eventID

func SubscribeCallback

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

Subscribe to messages matching query and launch a goroutine to run a callback for each one. The goroutine will exit if the callback returns true for 'stop' and clean up the subscription and channel.

Types

type Cache

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

A Cache buffers events for a Publisher.

func NewCache

func NewCache() *Cache

Create a new Cache with an EventSwitch as backend

func (*Cache) Flush

func (evc *Cache) Flush(publisher Publisher) error

func (*Cache) Publish

func (evc *Cache) Publish(ctx context.Context, message interface{}, tags Tags) error

Cache an event to be fired upon finality.

func (*Cache) Reset

func (evc *Cache) Reset()

func (*Cache) Sync

func (evc *Cache) Sync(publisher Publisher) error

Clears cached events by flushing them to Publisher

type CombinedTags

type CombinedTags []Tags

func (CombinedTags) Get

func (ct CombinedTags) Get(key string) (value string, ok bool)

func (CombinedTags) Keys

func (ct CombinedTags) Keys() []string

func (CombinedTags) Len

func (ct CombinedTags) Len() (length int)

func (CombinedTags) Map

func (ct CombinedTags) Map() map[string]interface{}

type Emitter

type Emitter interface {
	Subscribable
	Publisher
	process.Process
}

func NewEmitter

func NewEmitter(logger *logging.Logger) Emitter

type Indexable

type Indexable interface {
	ProvideIndex(index uint64)
}

If message implement this interface we will provide them with an index in the cache

type Publisher

type Publisher interface {
	Publish(ctx context.Context, message interface{}, tag Tags) error
}

func NewNoOpPublisher

func NewNoOpPublisher() Publisher

NoOpPublisher

type PublisherFunc

type PublisherFunc func(ctx context.Context, message interface{}, tags Tags) error

func (PublisherFunc) Publish

func (pf PublisherFunc) Publish(ctx context.Context, message interface{}, tags Tags) error

type Subscribable

type Subscribable interface {
	// Subscribe to all events matching query, which is a valid tmlibs Query. Blocking the out channel blocks the entire
	// pubsub.
	Subscribe(ctx context.Context, subscriber string, queryable query.Queryable, out chan<- interface{}) error
	// Unsubscribe subscriber from a specific query string. Note the subscribe channel must be drained.
	Unsubscribe(ctx context.Context, subscriber string, queryable query.Queryable) error
	UnsubscribeAll(ctx context.Context, subscriber string) error
}

TODO: manage the creation, closing, and draining of channels behind the interface rather than only closing. stop one subscriber from blocking everything!

type TagMap

type TagMap map[string]interface{}

func (TagMap) Get

func (ts TagMap) Get(key string) (value string, ok bool)

func (TagMap) Keys

func (ts TagMap) Keys() []string

func (TagMap) Len

func (ts TagMap) Len() int

func (TagMap) Map

func (ts TagMap) Map() map[string]interface{}

type Tags

type Tags interface {
	pubsub.TagMap
	Keys() []string
}