Documentation
¶
Overview ¶
Package fanout provides a Postgres-backed implementation of core/fanout.Fanout using LISTEN/NOTIFY.
PostgresFanout uses ONE channel, named after the fallback table ("gofastr_fanout_msgs" by default, so two fanouts with different tables on one database are fully isolated); the fanout topic travels inside the NOTIFY payload so dynamic topic strings never need to be validated as PG channel identifiers. Payloads that fit under the ~7000-byte inline threshold (leaving margin below Postgres's 8000-byte NOTIFY limit) are delivered inline; larger payloads are INSERTed into a fallback table and a short "t:<id>" pointer is NOTIFY'd, with receivers SELECTing the row by id.
Delivery is lossy best-effort, like every Fanout: pq.Listener reconnects automatically after a connection gap, but messages published during the gap are lost (the durable lane is the transactional outbox's job).
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*PostgresFanout)
Option configures a PostgresFanout.
func WithListenTimeout ¶
WithListenTimeout overrides the deadline NewPostgres waits for the initial LISTEN connection to become ready (default 15s). pq.Listener retries a bad or unroutable DSN forever; this deadline turns that into a prompt, descriptive construction error instead of an indefinite hang. A timeout closes the listener and aborts construction.
func WithTableName ¶
WithTableName overrides the fallback table name (default "gofastr_fanout_msgs"). The name must be a valid SQL identifier.
func WithoutEnsureTable ¶
func WithoutEnsureTable() Option
WithoutEnsureTable suppresses the CREATE TABLE IF NOT EXISTS that New otherwise runs at construction, mirroring the outbox's option of the same name. You must then create the fallback table via your own migration pipeline before publishing a large payload.
type PostgresFanout ¶
type PostgresFanout struct {
// contains filtered or unexported fields
}
PostgresFanout is a lossy-best-effort cfanout.Fanout backed by Postgres LISTEN/NOTIFY. One connection LISTENs on a single channel; topic routing is inside the payload. Large payloads spill to a fallback table and are re-fetched by id on receive.
func NewPostgres ¶
NewPostgres creates a Postgres-backed fanout. dsn is used for the LISTEN connection (pq.NewListener); db is used for NOTIFY sends and the fallback table. Construction creates the fallback table unless WithoutEnsureTable is passed.
func (*PostgresFanout) Close ¶
func (p *PostgresFanout) Close() error
Close tears down the listener, the dispatch goroutine, and every subscriber queue — including those whose cancel func the caller dropped, which would otherwise park their goroutines forever. Subscribe refuses new registrations after Close (it checks closed under the same mu, so no subscriber can slip in between the stop sweep and the flag).
func (*PostgresFanout) Publish ¶
Publish broadcasts payload to all subscribers of topic. Inline when the wrapped message fits under the threshold; otherwise spills to the fallback table and NOTIFYs a short pointer. Best-effort.
payload and topic must be valid UTF-8: NOTIFY payloads and the fallback table's TEXT column carry JSON envelopes, and Postgres would otherwise silently substitute U+FFFD for invalid bytes, corrupting the message.
func (*PostgresFanout) Subscribe ¶
func (p *PostgresFanout) Subscribe(topic string, fn func(payload []byte)) (cancel func(), err error)
Subscribe registers fn for topic. fn is wrapped in a per-subscriber bounded queue (cfanout.SubscriberQueue) so it runs on a dedicated goroutine with drop-oldest overflow — honoring the cfanout.Fanout contract that a slow subscriber is dropped, not backpressured, and never stalls other topics. The returned cancel unregisters fn and stops the goroutine; safe to call multiple times.
Source Files