events

Documentation

Index

• Constants
• Variables
• type DataStreamReceiver
• type Event
• 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) (context.CancelFunc, error)
• type Source
  • func NewStreamSource(eventID ID, stream rpcc.Stream, receiver DataStreamReceiver) Source
• type SourceFactory
  • func NewStreamSourceFactory(eventID ID, factory StreamFactory, receiver DataStreamReceiver) SourceFactory
• 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 NewCallWaitTask(ec *eval.Runtime, fn *eval.CompiledFunction, polling time.Duration) *WaitTask
  • func NewEvalWaitTask(ec *eval.Runtime, fn *eval.Function, polling time.Duration) *WaitTask
  • func NewValueWaitTask(when drivers.WaitEvent, value core.Value, getter Function, ...) *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")
)

Types

type DataStreamReceiver

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

type Event

type Event struct {
	ID   ID
	Data interface{}
}

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

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) (context.CancelFunc, 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,
	receiver DataStreamReceiver,
) Source

NewStreamSource create a new custom event source based on rpcc.Stream eventID - is a unique event ID stream - is a custom event stream receiver - 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 DataStreamReceiver) SourceFactory

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 NewCallWaitTask

func NewCallWaitTask(
	ec *eval.Runtime,
	fn *eval.CompiledFunction,
	polling time.Duration,
) *WaitTask

func NewEvalWaitTask

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

func NewValueWaitTask

func NewValueWaitTask(
	when drivers.WaitEvent,
	value core.Value,
	getter 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)