Documentation
¶
Index ¶
- Constants
- func AttachFanout(bus *EventBus, f fanout.Fanout) (stop func(), err error)
- func IsRemote(ctx context.Context) bool
- type Event
- type EventBus
- func (eb *EventBus) Emit(ctx context.Context, event Event) error
- func (eb *EventBus) EmitAsync(ctx context.Context, event Event)
- func (eb *EventBus) EmitStrict(ctx context.Context, event Event) error
- func (eb *EventBus) On(eventType string, handler EventHandler)
- func (eb *EventBus) Snapshot(eventType string) []EventHandler
- func (eb *EventBus) Subscribe(eventType string, handler EventHandler) (cancel func())
- type EventHandler
Constants ¶
const ( EntityCreated = "entity.created" EntityUpdated = "entity.updated" EntityDeleted = "entity.deleted" )
Pre-defined event types for entity lifecycle.
Variables ¶
This section is empty.
Functions ¶
func AttachFanout ¶ added in v0.16.0
AttachFanout bridges bus to f: every locally-emitted event (via Emit, EmitStrict, or EmitAsync) is published to the fanout so other replicas see it, and events published by OTHER replicas are re-emitted on this bus. With a fanout attached, On/Subscribe handlers fire on EVERY replica.
Delivery is lossy best-effort (the real-time lane); durable delivery is the outbox's job. Each event is wrapped in a node-id envelope; messages that originated on this bus are dropped on receive so broadcasts do not echo back. Remote re-emits run through EmitAsync with the tap suppressed (withRemoteReemit) so they are not re-published — preventing loops.
The tap NEVER blocks a synchronous emitter: it enqueues to a bounded queue serviced by a single publisher goroutine (drop-oldest on overflow), so a stalled fanout backend cannot stall Emit/EmitStrict.
DERIVED EVENTS: handlers that react to an event by emitting a NEW event must gate the derivation on IsRemote so it runs only on the origin replica — `if IsRemote(ctx) { return nil }`. Without the gate, every replica derives its own copy and remote replicas observe duplicates (the reaction handler runs on every replica, each re-deriving).
Event.Data round-trips through JSON: remote handlers see map[string]any for object payloads (crud events already are).
At most one fanout may be attached per bus; a second AttachFanout returns an error (detach via stop first). The returned stop detaches the tap, cancels the subscription, and stops the publisher goroutine; safe to call multiple times.
func IsRemote ¶ added in v0.16.0
IsRemote reports whether the event being handled arrived from another replica via an attached fanout. Handlers that DERIVE new events from the events they receive must gate on it — `if event.IsRemote(ctx) { return nil }` — so the derivation runs only on the origin replica; otherwise every replica derives its own copy and remote replicas observe duplicates.
Locally-emitted events (Emit/EmitAsync/EmitStrict on a normal context) report false; events re-emitted by AttachFanout from a remote replica report true.
Types ¶
type Event ¶
type Event struct {
// ID uniquely identifies this delivery. It is stamped by the
// outbox Relay (framework/outbox) from the persisted row's ID so
// consumers can deduplicate at-least-once deliveries. It is empty
// for events emitted directly via Emit/EmitAsync, which carry no
// durable identity.
ID string `json:"id,omitempty"`
Type string `json:"type"`
Data any `json:"data,omitempty"`
Timestamp time.Time `json:"timestamp"`
}
Event represents something that happened in the system.
type EventBus ¶
type EventBus struct {
// contains filtered or unexported fields
}
EventBus provides in-process publish/subscribe event delivery.
Both On (no cancel) and Subscribe (returns cancel) are supported. Handlers are stored in registration order so Emit short-circuits deterministically on the first error.
func (*EventBus) Emit ¶
Emit publishes an event synchronously to all subscribers and returns the first error from a handler.
A handler that panics is recovered so a buggy subscriber can't bring down the request that triggered the event. The panic is swallowed (the event bus is fire-and-iterate, not a critical-path return channel); callers that need hard failures should propagate them through the returned error instead. A nil entry — which shouldn't happen given Subscribe's guard, but kept for defense-in-depth against direct map manipulation — is skipped.
func (*EventBus) EmitAsync ¶
EmitAsync publishes an event in a goroutine (fire-and-forget).
Each handler runs inside emitSafe so a panicking subscriber takes itself down without crashing the process. Nil entries are skipped defensively.
func (*EventBus) EmitStrict ¶ added in v0.15.0
EmitStrict publishes synchronously like Emit, but treats a panicking subscriber as a delivery ERROR (returned to the caller) rather than swallowing it. Emit's swallow protects user transactions when the bus is wired into AfterCreate/AfterUpdate hooks; the transactional outbox relay has the opposite need — a consumer that panics must be retried and eventually dead-lettered, never silently marked dispatched — so it calls
func (*EventBus) On ¶
func (eb *EventBus) On(eventType string, handler EventHandler)
On subscribes a handler to the given event type. The handler stays registered for the lifetime of the bus; use Subscribe instead when you need to remove the handler later.
func (*EventBus) Snapshot ¶
func (eb *EventBus) Snapshot(eventType string) []EventHandler
Snapshot returns a copy of the handlers registered for the event type, in registration order, so emission doesn't hold the lock while user code runs.
func (*EventBus) Subscribe ¶
func (eb *EventBus) Subscribe(eventType string, handler EventHandler) (cancel func())
Subscribe registers a handler and returns a cancel function. Calling cancel removes the handler. Cancel is safe to call multiple times.
A nil handler is refused — the bus never retains a nil subscription, so a later Emit can't crash by invoking nothing. The returned cancel is a no-op in that case.
Source Files