events

Documentation

Overview

Package events contains interfaces for managing events within distributed systems

Index

• Variables
• type Event
  • func (e *Event) Unmarshal(v interface{}) error
• type PublishOption
  • func WithMetadata(md map[string]string) PublishOption
  • func WithPayload(p interface{}) PublishOption
  • func WithTimestamp(t time.Time) PublishOption
• type PublishOptions
• type ReadOption
  • func ReadFilter(key, value string) ReadOption
  • func ReadLimit(l int) ReadOption
  • func ReadOffset(l int) ReadOption
  • func ReadTopic(t string) ReadOption
• type ReadOptions
• type Store
• type Stream
• type SubscribeOption
  • func WithQueue(q string) SubscribeOption
  • func WithStartAtTime(t time.Time) SubscribeOption
  • func WithTopic(t string) SubscribeOption
• type SubscribeOptions
• type WriteOption
  • func WithTTL(d time.Duration) WriteOption
• type WriteOptions

Variables

var (
	// ErrMissingTopic is returned if a blank topic was provided to publish
	ErrMissingTopic = errors.New("Missing topic")
	// ErrEncodingMessage is returned from publish if there was an error encoding the message option
	ErrEncodingMessage = errors.New("Error encoding message")
)

Types

type Event

type Event struct {
	// ID to uniquely identify the event
	ID string
	// Topic of event, e.g. "registry.service.created"
	Topic string
	// Timestamp of the event
	Timestamp time.Time
	// Metadata contains the encoded event was indexed by
	Metadata map[string]string
	// Payload contains the encoded message
	Payload []byte
}

Event is the object returned by the broker when you subscribe to a topic

func (*Event) Unmarshal

func (e *Event) Unmarshal(v interface{}) error

Unmarshal the events message into an object

type PublishOption

type PublishOption func(o *PublishOptions)

PublishOption sets attributes on PublishOptions

func WithMetadata

func WithMetadata(md map[string]string) PublishOption

WithMetadata sets the Metadata field on PublishOptions

func WithPayload

func WithPayload(p interface{}) PublishOption

WithPayload sets the payload field on PublishOptions

func WithTimestamp

func WithTimestamp(t time.Time) PublishOption

WithTimestamp sets the timestamp field on PublishOptions

type PublishOptions

type PublishOptions struct {
	// Metadata contains any keys which can be used to query the data, for example a customer id
	Metadata map[string]string
	// Payload contains any additonal data which is relevent to the event but does not need to be
	// indexed such as structured data
	Payload interface{}
	// Timestamp to set for the event, if the timestamp is a zero value, the current time will be used
	Timestamp time.Time
}

PublishOptions contains all the options which can be provided when publishing an event

type ReadOption

type ReadOption func(o *ReadOptions)

ReadOption sets attributes on ReadOptions

func ReadFilter

func ReadFilter(key, value string) ReadOption

ReadFilter sets a key and value in the query

func ReadLimit

func ReadLimit(l int) ReadOption

ReadLimit sets the limit attribute on ReadOptions

func ReadOffset

func ReadOffset(l int) ReadOption

ReadOffset sets the offset attribute on ReadOptions

func ReadTopic

func ReadTopic(t string) ReadOption

ReadTopic sets the topic attribute on ReadOptions

type ReadOptions

type ReadOptions struct {
	// Topic to read events from, if no topic is provided events from all topics will be returned
	Topic string
	// Query to filter the results using. The store will query the metadata provided when the event
	// was written to the store
	Query map[string]string
	// Limit the number of results to return
	Limit int
	// Offset the results by this number, useful for paginated queries
	Offset int
}

ReadOptions contains all the options which can be provided when reading events from a store

type Store

type Store interface {
	Read(opts ...ReadOption) ([]*Event, error)
	Write(event *Event, opts ...WriteOption) error
}

Store of events

type Stream

type Stream interface {
	Publish(topic string, opts ...PublishOption) error
	Subscribe(opts ...SubscribeOption) (<-chan Event, error)
}

Stream of events

type SubscribeOption

type SubscribeOption func(o *SubscribeOptions)

SubscribeOption sets attributes on SubscribeOptions

func WithQueue

func WithQueue(q string) SubscribeOption

WithQueue sets the Queue fielf on SubscribeOptions to the value provided

func WithStartAtTime

func WithStartAtTime(t time.Time) SubscribeOption

WithStartAtTime sets the StartAtTime field on SubscribeOptions to the value provided

func WithTopic

func WithTopic(t string) SubscribeOption

WithTopic sets the topic to subscribe to

type SubscribeOptions

type SubscribeOptions struct {
	// Queue is the name of the subscribers queue, if two subscribers have the same queue the message
	// should only be published to one of them
	Queue string
	// Topic to subscribe to, if left blank the consumer will be subscribed to the firehouse topic which
	// recieves all events
	Topic string
	// StartAtTime is the time from which the messages should be consumed from. If not provided then
	// the messages will be consumed starting from the moment the Subscription starts.
	StartAtTime time.Time
}

SubscribeOptions contains all the options which can be provided when subscribing to a topic

type WriteOption

type WriteOption func(o *WriteOptions)

WriteOption sets attributes on WriteOptions

func WithTTL

func WithTTL(d time.Duration) WriteOption

WithTTL sets the TTL attribute on WriteOptions

type WriteOptions

type WriteOptions struct {
	// TTL is the duration the event should be recorded for, a zero value TTL indicates the event should
	// be stored indefinately
	TTL time.Duration
}

WriteOptions contains all the options which can be provided when writing an event to a store