events

type Decoder ¶ added in v0.16.0

type Decoder func(ctx context.Context, stream rpcc.Stream) (core.Value, error)

type Event ¶

type Event struct {
	ID   ID
	Data interface{}
}

Event represents a system event that is returned from an event source

type EventStream ¶ added in v0.16.0

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

func (e *EventStream) Close(_ context.Context) error

func (e *EventStream) Read(ctx context.Context) <-chan events.Message

type Factory ¶ added in v0.16.0

type Factory func(ctx context.Context) (rpcc.Stream, error)

type Function ¶

type Function func(ctx context.Context) (core.Value, error)

type Handler ¶ added in v0.10.0

type Handler func(ctx context.Context, message interface{}) bool

Handler represents a function that is called when a particular event occurs Returned boolean value indicates whether the handler needs to be called again False value indicated that it needs to be removed and never called again

func Always(fn func(ctx context.Context, message interface{})) Handler

func Once(fn func(ctx context.Context, message interface{})) Handler

type ID ¶ added in v0.10.0

type ID int

ID represents a unique event ID

func New(name string) ID

type Listener ¶ added in v0.10.0

type Listener struct {
	ID      ListenerID
	EventID ID
	Handler Handler
}

Listener is an internal listener representation

type ListenerID ¶ added in v0.10.0

type ListenerID int

ListenerID is an internal listener ID that can be used to unsubscribe from a particular event

type Loop ¶ added in v0.10.0

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

func NewLoop(sources ...SourceFactory) *Loop

func (loop *Loop) AddListener(eventID ID, handler Handler) ListenerID

func (loop *Loop) Listeners(eventID ID) int

func (loop *Loop) RemoveListener(eventID ID, listenerID ListenerID)

func (loop *Loop) Run(ctx context.Context) error

type Source ¶ added in v0.10.0

type Source interface {
	rpcc.Stream
	Recv() (Event, error)
}

Source represents a custom source of system events

func NewStreamSource(
	eventID ID,
	stream rpcc.Stream,
	decoder StreamDecoder,
) Source

type SourceFactory ¶ added in v0.16.0

type SourceFactory func(ctx context.Context) (Source, error)

SourceFactory represents a function that creates a new instance of Source.

func NewStreamSourceFactory(eventID ID, factory StreamFactory, receiver StreamDecoder) SourceFactory

type StreamDecoder ¶ added in v0.16.0

type StreamDecoder func(stream rpcc.Stream) (interface{}, error)

type StreamFactory ¶ added in v0.16.0

type StreamFactory func(ctx context.Context) (rpcc.Stream, error)

type StreamSource ¶ added in v0.16.0

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

StreamSource represents a helper struct for generating custom event sources

func (src *StreamSource) Close() error

func (src *StreamSource) ID() ID

func (src *StreamSource) Ready() <-chan struct{}

func (src *StreamSource) Recv() (Event, error)

func (src *StreamSource) RecvMsg(m interface{}) error

type WaitTask ¶

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

func NewEvalWaitTask(
	ec *eval.Runtime,
	fn *eval.Function,
	polling time.Duration,
) *WaitTask

func NewWaitTask(
	fun Function,
	polling time.Duration,
) *WaitTask

func (task *WaitTask) Run(ctx context.Context) (core.Value, error)