eventstream

Documentation

Index

• Constants
• type EventStream
  • func NewEventStream[M any](initial M) *EventStream[M]
  • func (es *EventStream[M]) CloseSubs()
  • func (es *EventStream[M]) CompareAndSend(m M, fn func(new M, old M) bool)
  • func (es *EventStream[M]) Latest() M
  • func (es *EventStream[M]) Leave(ch chan M)
  • func (es *EventStream[M]) Send(m M)
  • func (es *EventStream[M]) Shutdown()
  • func (es *EventStream[M]) Sub() chan M
  • func (es *EventStream[M]) SubStream(ctx context.Context) Stream[M]
• type Stream
  • func NewStream[T any](ctx context.Context, p *EventStream[T]) Stream[T]

Constants

const (
	SUBSCRIBE cmd = iota
	SEND
	SEND_COMPARE
	LEAVE
	SHUTDOWN
	CLOSE
	LENGTH
)

const TIMEOUT = time.Millisecond * 100

Types

type EventStream

type EventStream[M any] struct {
	// contains filtered or unexported fields
}

func NewEventStream

func NewEventStream[M any](initial M) *EventStream[M]

func (*EventStream[M]) CloseSubs

func (es *EventStream[M]) CloseSubs()

CloseSubs closes all channels handed out by Sub() and prevents new subs from subscribing

func (*EventStream[M]) CompareAndSend

func (es *EventStream[M]) CompareAndSend(m M, fn func(new M, old M) bool)

CompareAndSend sends the value m to all subscribers if the fn given returns true

func (*EventStream[M]) Latest

func (es *EventStream[M]) Latest() M

func (*EventStream[M]) Leave

func (es *EventStream[M]) Leave(ch chan M)

Leave leaves the subscriber list, the channel should be one returned by Sub, the channel is closed once the Leave request has been processed

func (*EventStream[M]) Send

func (es *EventStream[M]) Send(m M)

Send sends the value M to all subscribers previously subscribed through Sub() or SubStream(), the last value Send is also stored and send when a new subscriber appears.

func (*EventStream[M]) Shutdown

func (es *EventStream[M]) Shutdown()

Shutdown is like CloseSubs but also exits the background goroutine used for updating the eventstream

func (*EventStream[M]) Sub

func (es *EventStream[M]) Sub() chan M

Sub subscribers to this stream of events, the channel will receive values send through calling Send with a small buffer and receive grace period if they fall behind.

func (*EventStream[M]) SubStream

func (es *EventStream[M]) SubStream(ctx context.Context) Stream[M]

SubStream is like Sub but returns a Stream interface instead of a channel

type Stream

type Stream[T any] interface {
	Next() (T, error)
	Close() error
}

func NewStream

func NewStream[T any](ctx context.Context, p *EventStream[T]) Stream[T]