Documentation
¶
Overview ¶
Package events dispatches typed events to listeners registered by type. Listeners are plain functions; registration is explicit code (no discovery):
type UserRegistered struct{ User models.User }
bus.Register(events.On(listeners.SendWelcome)) // func(ctx, UserRegistered) error
err := bus.Dispatch(ctx, UserRegistered{User: u})
Async listeners run on a bounded goroutine pool that Shutdown drains. Queued listeners (OnQueued) run as queue jobs, so they survive restarts and are retried. Apps get a Bus in app.Events; Fake records dispatched events in tests.
See https://github.com/AlexGalhardo/vukem/blob/main/docs/subsystems.md#events.
Example ¶
package main
import (
"context"
"fmt"
"log/slog"
"github.com/AlexGalhardo/vukem/events"
)
type UserRegistered struct{ Email string }
func main() {
ctx := context.Background()
bus := events.New(slog.New(slog.DiscardHandler), 4)
defer func() { _ = bus.Shutdown(ctx) }()
bus.Register(events.On(func(_ context.Context, e UserRegistered) error {
fmt.Println("welcome mail to", e.Email)
return nil
}))
err := bus.Dispatch(ctx, UserRegistered{Email: "ada@example.com"})
fmt.Println(err)
}
Output: welcome mail to ada@example.com <nil>
Index ¶
- Variables
- func Dispatched[E any](f *Fake) []E
- func Listen[E any](b *Bus, fn func(ctx context.Context, e E) error)
- func ListenAsync[E any](b *Bus, fn func(ctx context.Context, e E) error)
- type Bus
- func (b *Bus) Dispatch(ctx context.Context, event any) error
- func (b *Bus) Jobs() []queue.Registration
- func (b *Bus) RecordTo(f *Fake)
- func (b *Bus) Register(regs ...Registration)
- func (b *Bus) SetObserver(obs Observer)
- func (b *Bus) SetQueue(fn func() (Enqueuer, error))
- func (b *Bus) Shutdown(ctx context.Context) error
- type Dispatcher
- type Enqueuer
- type Event
- type Fake
- type Observer
- type Registration
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrClosed = errors.New("events: bus is shut down")
ErrClosed is returned by Dispatch after Shutdown.
var ErrNoQueue = errors.New("events: queued listener needs a queue (Bus.SetQueue)")
ErrNoQueue is returned when a queued listener runs on a bus without a queue.
Functions ¶
func Dispatched ¶
Dispatched returns the recorded events of type E.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus holds listeners and runs them.
func (*Bus) Dispatch ¶
Dispatch runs the synchronous listeners of event's type in registration order (stopping at the first error) and schedules the async ones. Events without listeners are fine.
func (*Bus) Jobs ¶ added in v0.16.0
func (b *Bus) Jobs() []queue.Registration
Jobs returns the queue handlers of the queued listeners, to register on the queue.
func (*Bus) RecordTo ¶ added in v0.28.0
RecordTo makes Dispatch record every event in f instead of running the listeners (sync, async and queued), for tests that assert what was dispatched; read them with Dispatched. RecordTo(nil) restores normal dispatch.
func (*Bus) Register ¶
func (b *Bus) Register(regs ...Registration)
Register adds listeners in order.
func (*Bus) SetObserver ¶ added in v0.19.0
SetObserver reports dispatches to obs (development tooling such as devtools).
type Dispatcher ¶
Dispatcher dispatches events (Bus and Fake implement it).
type Enqueuer ¶ added in v0.16.0
type Enqueuer interface {
DispatchKind(ctx context.Context, kind string, payload any, opts ...queue.Option) error
}
Enqueuer sends events to queued listeners (*queue.Queue implements it).
type Event ¶ added in v0.19.0
type Event struct {
// Type is the event's Go type ("main.UserRegistered").
Type string
// Listeners is the number of listeners (sync, async and queued).
Listeners int
// Duration covers the synchronous listeners and enqueueing.
Duration time.Duration
// Err is the error Dispatch returned.
Err error
}
Event describes one Bus.Dispatch for an Observer. It carries the event's Go type only, never its fields (they may hold user data).
type Fake ¶
type Fake struct {
// contains filtered or unexported fields
}
Fake records dispatched events instead of running listeners (tests).
Example ¶
package main
import (
"context"
"fmt"
"github.com/AlexGalhardo/vukem/events"
)
type UserRegistered struct{ Email string }
func main() {
fake := events.NewFake()
_ = fake.Dispatch(context.Background(), UserRegistered{Email: "ada@example.com"})
for _, e := range events.Dispatched[UserRegistered](fake) {
fmt.Println("dispatched", e.Email)
}
}
Output: dispatched ada@example.com
type Observer ¶ added in v0.19.0
Observer is notified after every Bus.Dispatch. It runs on the caller's goroutine, so it must be fast and safe for concurrent use.
type Registration ¶
type Registration struct {
// contains filtered or unexported fields
}
Registration binds a listener to its event type; create it with On, OnAsync or OnQueued.
func On ¶
func On[E any](fn func(ctx context.Context, e E) error) Registration
On registers fn for events of type E, run synchronously by Dispatch (its error is returned to the caller).
func OnAsync ¶
func OnAsync[E any](fn func(ctx context.Context, e E) error) Registration
OnAsync registers fn for events of type E, run in the background; its error is logged.
func OnQueued ¶ added in v0.16.0
func OnQueued[E any](kind string, fn func(ctx context.Context, e E) error, opts ...queue.Option) Registration
OnQueued registers fn for events of type E as the queue job kind: Dispatch enqueues the event (encoded as JSON) and a queue worker runs fn with the queue's retries. The bus needs a queue (SetQueue; vukem.App wires App.Queue) and the job handler must be registered on it (Jobs).
Source Files