event

Documentation

Overview

Package event provides the event and its payload emitted by the user of a mesh and the contained cells. It is read-only to avoid modifications while multiple behaviors are using it concurrently. Accessors and cloners make live for behavior developers as well easier as the sink and its methods for analyzing.

Index

• Constants
• func NewReplyPayload(kvs ...interface{}) (*Payload, PayloadChan)
• type CriterionMatch
• type Event
  • func New(topic string, kvs ...interface{}) *Event
  • func WithContext(ctx context.Context, topic string, kvs ...interface{}) *Event
  • func (e *Event) Context() context.Context
  • func (e *Event) Done() bool
  • func (e *Event) Payload() *Payload
  • func (e *Event) Timestamp() time.Time
  • func (e *Event) Topic() string
• type Payload
  • func NewPayload(kvs ...interface{}) *Payload
  • func (pl *Payload) At(keys ...string) *Value
  • func (pl *Payload) Clone(kvs ...interface{}) *Payload
  • func (pl *Payload) Do(f func(key string, value *Value) error) error
  • func (pl *Payload) Keys() []string
  • func (pl *Payload) Len() int
  • func (pl *Payload) Reply(rpl *Payload) error
  • func (pl *Payload) String() string
• type PayloadChan
  • func (plc PayloadChan) Wait(timeout time.Duration) (*Payload, error)
• type Sink
  • func NewCheckedSink(max int, checker SinkProcessor) *Sink
  • func NewSink(max int) *Sink
  • func (s *Sink) Clear() error
  • func (s *Sink) Do(doer SinkDoer) error
  • func (s *Sink) Len() int
  • func (s *Sink) PeekAt(index int) (*Event, bool)
  • func (s *Sink) PeekFirst() (*Event, bool)
  • func (s *Sink) PeekLast() (*Event, bool)
  • func (s *Sink) PullFirst() (*Event, error)
  • func (s *Sink) PullLast() (*Event, error)
  • func (s *Sink) Push(evt *Event) (int, error)
• type SinkAccessor
• type SinkAnalyzer
  • func NewSinkAnalyzer(accessor SinkAccessor) *SinkAnalyzer
  • func (sa *SinkAnalyzer) Filter(filter SinkFilter) (SinkAccessor, error)
  • func (sa *SinkAnalyzer) Fold(inject *Payload, folder SinkFolder) (*Payload, error)
  • func (sa *SinkAnalyzer) Match(matcher SinkFilter) (bool, error)
  • func (sa *SinkAnalyzer) MinMaxDuration() (time.Duration, time.Duration)
  • func (sa *SinkAnalyzer) TopicFolds(folder SinkFolder) (map[string]*Payload, error)
  • func (sa *SinkAnalyzer) TopicQuantities() map[string]int
  • func (sa *SinkAnalyzer) TotalDuration() time.Duration
• type SinkDoer
• type SinkFilter
• type SinkFolder
• type SinkProcessor
• type Value
  • func (v *Value) AsBool(dv bool) bool
  • func (v *Value) AsDuration(dv time.Duration) time.Duration
  • func (v *Value) AsFloat64(dv float64) float64
  • func (v *Value) AsInt(dv int) int
  • func (v *Value) AsPayload() *Payload
  • func (v *Value) AsPayloadChan() PayloadChan
  • func (v *Value) AsString(dv string) string
  • func (v *Value) AsTime(dv time.Time) time.Time
  • func (v *Value) Error() string
  • func (v *Value) IsDefined() bool
  • func (v *Value) IsPayload() bool
  • func (v *Value) IsUndefined() bool
  • func (v *Value) String() string

Constants

const (
	TopicCollected = "collected"
	TopicCounted   = "counted"
	TopicProcess   = "process"
	TopicProcessed = "processed"
	TopicReset     = "reset"
	TopicResult    = "result"
	TopicStatus    = "status"
	TopicTick      = "tick"
)

Standard topics.

const (
	DefaultKey   = "value"
	DefaultValue = true
)

Defaults.

Functions

func NewReplyPayload

func NewReplyPayload(kvs ...interface{}) (*Payload, PayloadChan)

NewReplyPayload creates a new payload like NewPayload() but also a reply channel.

Types

type CriterionMatch

type CriterionMatch int

CriterionMatch signals, how a criterion matches.

const (
	CriterionDone CriterionMatch = iota + 1
	CriterionKeep
	CriterionDropFirst
	CriterionDropLast
	CriterionClear
)

List of criterion match signals.

type Event

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

Event describes an event of the cells. It contains a topic as well as a possible number of key/value pairs as payload.

func New

func New(topic string, kvs ...interface{}) *Event

New creates a new event. The arguments after the topic are taken to create a new payload.

func WithContext

func WithContext(ctx context.Context, topic string, kvs ...interface{}) *Event

WithContext creates a new event containing a context allowing to cancel it. The arguments after the topic are taken to create a new payload.

func (*Event) Context

func (e *Event) Context() context.Context

Context returns the event context.

func (*Event) Done

func (e *Event) Done() bool

Done tells if the event is done in the sense of the context. This happens if a potential context is canceled or reached timeout or deadline.

func (*Event) Payload

func (e *Event) Payload() *Payload

Payload returns the event payload.

func (*Event) Timestamp

func (e *Event) Timestamp() time.Time

Timestamp returns the event timestamp.

func (*Event) Topic

func (e *Event) Topic() string

Topic returns the event topic.

type Payload

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

Payload contains key/value pairs of data.

func NewPayload

func NewPayload(kvs ...interface{}) *Payload

NewPayload creates a payload with the given pairs of keys and values. In case of payloads or maps as keys thos will be merged, in case of arrays or slices those will be merged with the index as string. In case of those as values they will all be nested payload values.

func (*Payload) At

func (pl *Payload) At(keys ...string) *Value

At returns the value at the given key. This value may be empty.

func (*Payload) Clone

func (pl *Payload) Clone(kvs ...interface{}) *Payload

Clone creates a new payload with the content of the current one and applies the given changes.

func (*Payload) Do

func (pl *Payload) Do(f func(key string, value *Value) error) error

Do performs a function for all key/value pairs.

func (*Payload) Keys

func (pl *Payload) Keys() []string

Keys returns the keys of the payload.

func (*Payload) Len

func (pl *Payload) Len() int

Len returns the number of values of the payload.

func (*Payload) Reply

func (pl *Payload) Reply(rpl *Payload) error

Reply allows the receiver of a payload to reply via a channel.

func (*Payload) String

func (pl *Payload) String() string

String implements fmt.Stringer.

type PayloadChan

type PayloadChan chan *Payload

PayloadChan is intended to be sent with an event as payload so that a behavior can use it to answer a request.

func (PayloadChan) Wait

func (plc PayloadChan) Wait(timeout time.Duration) (*Payload, error)

Wait waits for a returned payload until receiving or timeout.

type Sink

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

Sink stores a number of events ordered by adding them at the end. To be used in behaviors for collecting sets of events and operate on them.

func NewCheckedSink

func NewCheckedSink(max int, checker SinkProcessor) *Sink

NewCheckedSink creates a sink for events.

func NewSink

func NewSink(max int) *Sink

NewSink creates a sink for events.

func (*Sink) Clear

func (s *Sink) Clear() error

Clear removes all collected events.

func (*Sink) Do

func (s *Sink) Do(doer SinkDoer) error

Do implements SinkAccessor.

func (*Sink) Len

func (s *Sink) Len() int

Len implements SinkAccessor.

func (*Sink) PeekAt

func (s *Sink) PeekAt(index int) (*Event, bool)

PeekAt implements SinkAccessor.

func (*Sink) PeekFirst

func (s *Sink) PeekFirst() (*Event, bool)

PeekFirst implements SinkAccessor.

func (*Sink) PeekLast

func (s *Sink) PeekLast() (*Event, bool)

PeekLast implements SinkAccessor.

func (*Sink) PullFirst

func (s *Sink) PullFirst() (*Event, error)

PullFirst returns and removed the first event of the sink.

func (*Sink) PullLast

func (s *Sink) PullLast() (*Event, error)

PullLast returns and removed the last event of the sink.

func (*Sink) Push

func (s *Sink) Push(evt *Event) (int, error)

Push adds a new event to the sink.

type SinkAccessor

type SinkAccessor interface {
	// Len returns the number of stored events.
	Len() int

	// PeekFirst returns the first of the collected events.
	PeekFirst() (*Event, bool)

	// PeekLast returns the last of the collected event datas.
	PeekLast() (*Event, bool)

	// PeekAt returns an event at a given index and true if it
	// exists, otherwise nil and false.
	PeekAt(index int) (*Event, bool)

	// Do iterates over all collected events.
	Do(doer SinkDoer) error
}

SinkAccessor can be used to read the events in a sink. It is a specialized subfunctionality of the event sink.

type SinkAnalyzer

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

SinkAnalyzer is a helpful type to analyze the events collected inside an event sink. It's intended to make the life for behavior developers more simple.

func NewSinkAnalyzer

func NewSinkAnalyzer(accessor SinkAccessor) *SinkAnalyzer

NewSinkAnalyzer creates an analyzer for the given sink accessor.

func (*SinkAnalyzer) Filter

func (sa *SinkAnalyzer) Filter(filter SinkFilter) (SinkAccessor, error)

Filter creates a new accessor containing only the filtered events.

func (*SinkAnalyzer) Fold

func (sa *SinkAnalyzer) Fold(inject *Payload, folder SinkFolder) (*Payload, error)

Fold reduces (folds) the events of the sink.

func (*SinkAnalyzer) Match

func (sa *SinkAnalyzer) Match(matcher SinkFilter) (bool, error)

Match checks if all events match the passed criterion.

func (*SinkAnalyzer) MinMaxDuration

func (sa *SinkAnalyzer) MinMaxDuration() (time.Duration, time.Duration)

MinMaxDuration returns the minimum and maximum durations between events in the sink.

func (*SinkAnalyzer) TopicFolds

func (sa *SinkAnalyzer) TopicFolds(folder SinkFolder) (map[string]*Payload, error)

TopicFolds reduces the events per topic.

func (*SinkAnalyzer) TopicQuantities

func (sa *SinkAnalyzer) TopicQuantities() map[string]int

TopicQuantities returns a map of collected topics and their quantity.

func (*SinkAnalyzer) TotalDuration

func (sa *SinkAnalyzer) TotalDuration() time.Duration

TotalDuration returns the duration between the first and the last event.

type SinkDoer

type SinkDoer func(index int, evt *Event) error

SinkDoer performs an operation on an event.

type SinkFilter

type SinkFilter func(index int, evt *Event) (bool, error)

SinkFilter checks if an event matches a criterium.

type SinkFolder

type SinkFolder func(index int, acc *Payload, evt *Event) (*Payload, error)

SinkFolder allows to reduce (fold) events.

type SinkProcessor

type SinkProcessor func(accessor SinkAccessor) (*Payload, error)

SinkProcessor can be used as a checker function but also inside of behaviors to process the content of an event sink and return a new payload.

type Value

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

Value contains one payload value.

func (*Value) AsBool

func (v *Value) AsBool(dv bool) bool

AsBool returns the value as bool, dv is taken as default value.

func (*Value) AsDuration

func (v *Value) AsDuration(dv time.Duration) time.Duration

AsDuration returns the value as duration, dv is taken as default value.

func (*Value) AsFloat64

func (v *Value) AsFloat64(dv float64) float64

AsFloat64 returns the value as float64, dv is taken as default value.

func (*Value) AsInt

func (v *Value) AsInt(dv int) int

AsInt returns the value as int, dv is taken as default value.

func (*Value) AsPayload

func (v *Value) AsPayload() *Payload

AsPayload returns the value as payload.

func (*Value) AsPayloadChan

func (v *Value) AsPayloadChan() PayloadChan

AsPayloadChan returns the value as payload channel.

func (*Value) AsString

func (v *Value) AsString(dv string) string

AsString returns the value as string, dv is taken as default value.

func (*Value) AsTime

func (v *Value) AsTime(dv time.Time) time.Time

AsTime returns the value as time, dv is taken as default value.

func (*Value) Error

func (v *Value) Error() string

Error implements error.

func (*Value) IsDefined

func (v *Value) IsDefined() bool

IsDefined returns true if this value is defined.

func (*Value) IsPayload

func (v *Value) IsPayload() bool

IsPayload returns true if the value is a payload.

func (*Value) IsUndefined

func (v *Value) IsUndefined() bool

IsUndefined returns true if this value is undefined.

func (*Value) String

func (v *Value) String() string

String implements fmt.Stringer.