Documentation
¶
Overview ¶
Package fanout carries real-time messages between replicas.
GoFastr's real-time surfaces — [event.EventBus], [island.Manager], and [stream.SSEBroker] — are per-process by default. A write on replica A notifies only the browsers connected to A. Fanout is the seam that lets those surfaces broadcast across replicas.
Lossy best-effort by design ¶
This is the real-time lane. Implementations are permitted to drop a message published while a subscriber is disconnected, reconnecting, or queue-full. Durable, at-least-once delivery is the transactional outbox's job (framework/outbox — see "Two delivery lanes"). The two lanes are disjoint: fanout never participates in the durable path.
Loop guard ¶
Every node publishes its messages wrapped in an envelope that carries the originator's node id (Wrap). Receivers drop any message whose node id matches their own (Unwrap) so a broadcast is not re-broadcast. The integrations (bus bridge, island manager, SSE broker) apply this guard on receive; the Fanout itself is node-agnostic.
Backends ¶
InProcess — an in-memory pub/sub. Its primary purpose is tests: wiring two buses / two island managers / two brokers to one InProcess simulates two replicas inside a single test binary.
NewRedis — a thin adapter over a user-supplied RedisPubSub. No Redis library is imported; adapt go-redis (or redigo) in ~15 lines:
// go-redis adapter type goRedisPubSub struct{ c *redis.Client }
func (a goRedisPubSub) Publish(ctx context.Context, ch string, p []byte) error { return a.c.Publish(ctx, ch, p).Err() } func (a goRedisPubSub) Subscribe(ctx context.Context, ch string, fn func([]byte)) (func(), error) { sub := a.c.Subscribe(ctx, ch) ch2 := sub.Channel(redis.WithChannelSize(64)) done := make(chan struct{}) go func() { defer close(done); for m := range ch2 { fn([]byte(m.Payload)) } }() return func() { _ = sub.Close(); <-done }, nil } fanout.NewRedis(goRedisPubSub{c: client})
Trusted transport ¶
The fanout transport is trusted input: write access to the underlying channel (a Postgres NOTIFY channel, a Redis pub/sub channel, …) equals event-injection into every replica. Payloads are not authenticated — a forged envelope published to the channel is re-emitted on every bus that subscribes to it. Secure the channel (network isolation, DB/Redis credentials) rather than relying on the fanout to reject malicious input.
For a Postgres LISTEN/NOTIFY backend see package framework/fanout.
Index ¶
- func NewNodeID() string
- func PublishQueue(f Fanout, topic string, depth int) (send func([]byte), stop func())
- func SubscriberQueue(fn func([]byte), depth int) (send func([]byte), stop func())
- func Unwrap(raw []byte) (nodeID string, body []byte, err error)
- func Wrap(nodeID string, body []byte) []byte
- type Fanout
- type InProcess
- type InProcessOption
- type RedisPubSub
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewNodeID ¶
func NewNodeID() string
NewNodeID returns 16 random bytes hex-encoded (32 chars). Random rather than counter-based avoids both global-contention and the assumption that every replica is built from the same source — two replicas that happen to both start at counter zero would loop on each other's broadcasts.
crypto/rand.Read on the default Reader does not fail on a supported platform (getrandom/SecRandomCopyBytes/RtlGenRandom), so the error is ignored — the modern idiom used elsewhere in the framework (e.g. core/middleware request ids).
func PublishQueue ¶
PublishQueue returns a non-blocking send for mirroring messages to a Fanout from hot paths (HTTP handlers, event emitters). Publishing to a real backend is a network/DB round-trip; calling Fanout.Publish inline would hand every caller an unbounded stall on a slow or wedged backend.
send enqueues into a bounded, drop-oldest queue drained by one dedicated goroutine that publishes each payload under a fixed per-publish deadline; it never blocks and never returns an error — publish failures are logged at Debug (lossy lane: the durable path is the outbox's job). After stop, send is a silent no-op; stop is prompt and safe to call multiple times.
depth <= 0 selects the default queue depth.
func SubscriberQueue ¶
SubscriberQueue wraps a subscriber callback in a per-subscriber bounded queue with drop-oldest overflow, running fn on a dedicated goroutine.
It is an implementation aid for Fanout backends whose transport delivers payloads on a SHARED goroutine — e.g. a single LISTEN/NOTIFY dispatcher (framework/fanout) or a Redis reader goroutine (NewRedis). Wrapping each subscriber in its own queue preserves the per-subscriber bounded-queue + drop-oldest contract the Fanout.Subscribe doc promises, so one slow subscriber cannot stall delivery to the others (and never blocks the shared transport goroutine). InProcess uses it internally too.
send is safe to call concurrently and NEVER blocks: when the queue is full it drops the oldest queued payload and enqueues the new one. After stop, send is a silent no-op. stop signals the goroutine to exit and returns promptly; safe to call multiple times.
depth <= 0 selects the default queue depth.
func Unwrap ¶
Unwrap decodes an envelope produced by Wrap, returning the originator's nodeID and the original body. It errors if raw is not a valid envelope or carries an empty node id.
Types ¶
type Fanout ¶
type Fanout interface {
// Publish broadcasts payload to all subscribers of topic on every node.
// It is best-effort: a slow or absent subscriber is dropped, not
// backpressured (unless the backend cannot help it).
Publish(ctx context.Context, topic string, payload []byte) error
// Subscribe registers fn for every payload published on topic by any
// node. fn must not block; backends invoke it on a dedicated goroutine
// with a bounded queue and drop oldest on overflow (mirroring
// [stream.SSEBroker]). The returned cancel unregisters fn; safe to call
// multiple times.
Subscribe(topic string, fn func(payload []byte)) (cancel func(), err error)
}
Fanout carries real-time messages between replicas. Implementations are lossy best-effort: a message published while a subscriber is disconnected, reconnecting, or queue-full is gone. Durable delivery belongs to the transactional outbox (framework/outbox).
Payloads must be valid UTF-8: framework payloads are JSON envelopes, and a backend may reject invalid UTF-8 (e.g. the Postgres backend, which would otherwise silently corrupt such bytes via U+FFFD substitution). The fanout transport is trusted input — write access to the underlying channel equals event-injection into every replica (forged payloads are not authenticated).
func NewRedis ¶
func NewRedis(client RedisPubSub) Fanout
NewRedis returns a Fanout backed by the supplied RedisPubSub. The caller owns the Redis client and adapts it to RedisPubSub.
type InProcess ¶
type InProcess struct {
// contains filtered or unexported fields
}
InProcess is an in-memory Fanout. Its primary purpose is tests: wiring two buses / two island managers / two brokers to one InProcess simulates two replicas inside a single test binary. It is fully concurrency-safe and delivers per-subscriber in publish order.
func NewInProcess ¶
func NewInProcess(opts ...InProcessOption) *InProcess
NewInProcess returns a concurrency-safe in-memory fanout.
func (*InProcess) Publish ¶
Publish broadcasts payload to every subscriber of topic. It never blocks: a subscriber whose queue is full has its oldest queued message dropped to make room (mirroring [stream.SSEBroker]).
type InProcessOption ¶
type InProcessOption func(*InProcess)
InProcessOption configures an InProcess.
func WithInProcessQueue ¶
func WithInProcessQueue(depth int) InProcessOption
WithInProcessQueue overrides the per-subscriber bounded queue depth (default 256). A depth of 0 keeps the default. Mainly useful in tests that need to exercise the drop-oldest overflow path quickly.
type RedisPubSub ¶
type RedisPubSub interface {
Publish(ctx context.Context, channel string, payload []byte) error
Subscribe(ctx context.Context, channel string, fn func(payload []byte)) (cancel func(), err error)
}
RedisPubSub is the minimal Redis surface NewRedis needs. No Redis library is imported; implement this with your preferred client (go-redis, redigo, …). See the package doc for a go-redis adapter example.
Subscribe must invoke fn for every message published on channel by ANY client, including other processes/replicas. The returned cancel stops delivery and releases the subscription.
Source Files