events

Documentation

Index

• Constants
• Variables
• func NewEventStream(stream rpcc.Stream, decoder Decoder) events.Stream
• type Decoder
• type Event
• type EventStream
  • func (e *EventStream) Close(_ context.Context) error
  • func (e *EventStream) Read(ctx context.Context) <-chan events.Message
• type Factory
• type Function
• type Handler
  • func Always(fn func(ctx context.Context, message interface{})) Handler
  • func Once(fn func(ctx context.Context, message interface{})) Handler
• type ID
  • func New(name string) ID
• type Listener
• type ListenerID
• type Loop
  • 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
  • func NewStreamSource(eventID ID, stream rpcc.Stream, decoder StreamDecoder) Source
• type SourceFactory
  • func NewStreamSourceFactory(eventID ID, factory StreamFactory, receiver StreamDecoder) SourceFactory
• type StreamDecoder
• type StreamFactory
• type StreamSource
  • 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
  • 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)

Constants

const DefaultPolling = time.Millisecond * time.Duration(200)

Variables

var (
	Error = New("error")
)

Functions

func NewEventStream

func NewEventStream(stream rpcc.Stream, decoder Decoder) events.Stream

Types

type Decoder

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

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

func (*EventStream) Close

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

func (*EventStream) Read

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

type Factory

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

type Function

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

type Handler

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

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

Always returns a handler wrapper that always gets executed by an event loop

func Once

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

Once returns a handler wrapper that gets executed only once by an event loop

type ID

type ID int

ID represents a unique event ID

func New

func New(name string) ID

type Listener

type Listener struct {
	ID      ListenerID
	EventID ID
	Handler Handler
}

Listener is an internal listener representation

type ListenerID

type ListenerID int

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

type Loop

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

func NewLoop

func NewLoop(sources ...SourceFactory) *Loop

func (*Loop) AddListener

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

func (*Loop) Listeners

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

func (*Loop) RemoveListener

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

func (*Loop) Run

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

type Source

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

Source represents a custom source of system events

func NewStreamSource

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

NewStreamSource create a new custom event source based on rpcc.Stream eventID - is a unique event ID stream - is a custom event stream decoder - is a value conversion function

type SourceFactory

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

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

func NewStreamSourceFactory

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

type StreamDecoder

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

type StreamFactory

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

type StreamSource

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

StreamSource represents a helper struct for generating custom event sources

func (*StreamSource) Close

func (src *StreamSource) Close() error

func (*StreamSource) ID

func (src *StreamSource) ID() ID

func (*StreamSource) Ready

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

func (*StreamSource) Recv

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

func (*StreamSource) RecvMsg

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

type WaitTask

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

func NewEvalWaitTask

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

func NewWaitTask

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

func (*WaitTask) Run

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