event

Documentation

Overview

Package event provides in-memory event brokers.

Index

• func None[Event any](_ context.Context, _ Event) error
• func WithStreamQueueSize(size int32) func(options *streamOptions)
• type ContextMiddleware
• type Mux
  • func NewMux[Event any]() *Mux[Event]
  • func (m *Mux[Event]) Observe(ctx context.Context, event Event) error
  • func (m *Mux[Event]) Shutdown(wg *sync.WaitGroup)
  • func (m *Mux[Event]) WillNotify(observers ...Observer[Event]) *Mux[Event]
  • func (m *Mux[Event]) WithContextMiddleware(middleware ...ContextMiddleware) *Mux[Event]
  • func (m *Mux[Event]) WithObserverMiddleware(middleware ...ObserverMiddleware[Event]) *Mux[Event]
• type Observer
• type ObserverMiddleware
• type Stream
  • func NewStream[Event any](optionsFn ...func(options *streamOptions)) *Stream[Event]
  • func (s *Stream[Event]) Observe(ctx context.Context, e Event) error
  • func (s *Stream[Event]) Shutdown(wg *sync.WaitGroup)
  • func (s *Stream[Event]) WillNotify(consume Observer[Event]) *Stream[Event]
  • func (s *Stream[Event]) WithContextMiddleware(middleware ...ContextMiddleware) *Stream[Event]
  • func (s *Stream[Event]) WithObserverMiddleware(middleware ...ObserverMiddleware[Event]) *Stream[Event]

Functions

func None

func None[Event any](_ context.Context, _ Event) error

None is an event handler that does nothing.

If you have a service that supports listeners, inject None by default during construction, and you can stop checking for `nil` before notifying them.

Example:

// Construction.
h.listener = event.None[MyEvent]

// Usage (no need to check if `listener` is nil).
h.listener(ctx, e)

func WithStreamQueueSize

func WithStreamQueueSize(size int32) func(options *streamOptions)

Types

type ContextMiddleware

type ContextMiddleware func(ctx context.Context) context.Context

ContextMiddleware is used to transform the context provided to a function.

It exists as an optimization over ObserverMiddleware whenever the value or type of the event is irrelevant, which supports reuse without needlessly depending on the instantiation of generics.

type Mux

type Mux[Event any] struct {
	// contains filtered or unexported fields
}

Mux is a thread-safe in-memory event multiplexer.

func NewMux

func NewMux[Event any]() *Mux[Event]

NewMux creates a Mux.

func (*Mux[Event]) Observe

func (m *Mux[Event]) Observe(ctx context.Context, event Event) error

Observe and propagate an event to registered observers.

func (*Mux[Event]) Shutdown

func (m *Mux[Event]) Shutdown(wg *sync.WaitGroup)

Shutdown the Mux and communicate finishing via the sync.WaitGroup.

func (*Mux[Event]) WillNotify

func (m *Mux[Event]) WillNotify(
	observers ...Observer[Event],
) *Mux[Event]

WillNotify registers an observer. All events observed by Mux will be propagated to all registered observers.

func (*Mux[Event]) WithContextMiddleware

func (m *Mux[Event]) WithContextMiddleware(
	middleware ...ContextMiddleware,
) *Mux[Event]

WithContextMiddleware registers context middleware.

Context middleware will always be applied before observer middleware.

func (*Mux[Event]) WithObserverMiddleware

func (m *Mux[Event]) WithObserverMiddleware(
	middleware ...ObserverMiddleware[Event],
) *Mux[Event]

WithObserverMiddleware registers observer middleware.

Context middleware will always be applied before observer middleware.

type Observer

type Observer[Event any] func(ctx context.Context, e Event) error

Observer is a function that can process a specific event type.

The correct handling of errors depends on what the observer is:

• Event brokers, such as Mux, must indicate whether the propagation of the event succeeded, regardless of whether the observers succeeded.

• Middleware must propagate the errors they receive unless the purpose of that middleware is to modify the error.

• Asynchronous operations must return whether they succeeded. Brokers may depend on this to retry failed operations.

type ObserverMiddleware

type ObserverMiddleware[Event any] func(next Observer[Event]) Observer[Event]

ObserverMiddleware can be used to run arbitrary actions before and after the completion of next.

Note that there are two different ways in which we can use this within a Mux:

• With Mux.WithObserverMiddleware, which will apply it to all observers.
• By wrapping the observer before passing it to Mux.WillNotify, which will only apply to this particular observer.

While more involved, the latter approach is more powerful since it allows handling based on both the event and the observer. For example, logging middleware could attach an observer name or ID to each log, making them easier to understand.

type Stream

type Stream[Event any] struct {
	// contains filtered or unexported fields
}

Stream is a thread-safe in-memory event stream.

func NewStream

func NewStream[Event any](
	optionsFn ...func(options *streamOptions),
) *Stream[Event]

NewStream creates a Stream with the specified maximum queue size.

func (*Stream[Event]) Observe

func (s *Stream[Event]) Observe(ctx context.Context, e Event) error

Observe an event and propagate it to existing consumers.

This function never returns an error.

func (*Stream[Event]) Shutdown

func (s *Stream[Event]) Shutdown(wg *sync.WaitGroup)

Shutdown the Stream and communicate finishing via the sync.WaitGroup.

func (*Stream[Event]) WillNotify

func (s *Stream[Event]) WillNotify(consume Observer[Event]) *Stream[Event]

WillNotify and handle events.

Events produced before this function is called cannot be observed. The consume function runs in a new goroutine. This function never returns an error.

func (*Stream[Event]) WithContextMiddleware

func (s *Stream[Event]) WithContextMiddleware(
	middleware ...ContextMiddleware,
) *Stream[Event]

WithContextMiddleware registers context middleware.

Context middleware will always be applied before observer middleware.

func (*Stream[Event]) WithObserverMiddleware

func (s *Stream[Event]) WithObserverMiddleware(
	middleware ...ObserverMiddleware[Event],
) *Stream[Event]

WithObserverMiddleware registers observer middleware.

Context middleware will always be applied before observer middleware.