Documentation
¶
Overview ¶
Package postgres implements pubsub.Broker on PostgreSQL LISTEN/NOTIFY, so a horizontally scaled deployment that already uses Postgres for documents needs no extra infrastructure.
All topics share one notification channel (Options.Channel). Each message is a JSON envelope {"topic": "...", "payload": "<base64>"}. NOTIFY payloads are limited to 8000 bytes, so larger messages are written to the table backyard_pubsub_spill (id bigserial, payload bytea, created_at timestamptz) and the notification carries {"topic": "...", "spill_id": N} instead; the insert and the notification commit atomically, and receivers read the row back. A janitor deletes spill rows older than Options.SpillTTL. The table is created on startup in the first schema of the connection's search_path.
A dedicated connection runs LISTEN; when it fails, it is re-established with exponential backoff (notifications sent meanwhile are lost, matching the at-most-once contract). Idle listener connections are health-checked periodically.
Publishing, spill reads and the janitor share a small connection pool. Each of those is a single short statement, so the pool opened from DSN is bounded to Options.MaxPoolConns, by default derived from the CPU count and clamped to 2–4 connections (pgxpool's own default of max(4, NumCPU) would open dozens of connections per instance on large machines). A pool_max_conns parameter in the DSN takes precedence over both.
Index ¶
Constants ¶
const ( DefaultChannel = "backyard_events" DefaultSpillTTL = 5 * time.Minute DefaultJanitorInterval = time.Minute DefaultHealthInterval = 30 * time.Second )
Defaults for Options.
const SpillTable = "backyard_pubsub_spill"
SpillTable is the name of the table holding payloads too large for NOTIFY.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker is a PostgreSQL LISTEN/NOTIFY pubsub.Broker.
func New ¶
New connects to PostgreSQL, creates the spill table if needed and starts listening. It fails if the first LISTEN connection cannot be established.
func (*Broker) Close ¶
Close stops listening, stops all subscriptions and closes the pool (when the broker opened it). It is idempotent.
type Options ¶
type Options struct {
// DSN is a PostgreSQL connection string (URL or key=value form).
DSN string
// Channel is the NOTIFY channel shared by all topics (default
// "backyard_events"; at most 63 bytes).
Channel string
// Pool, if set, is used for publishing and spill reads instead of a pool
// opened from DSN. It is not closed by Broker.Close. DSN is still required
// for the dedicated LISTEN connection.
Pool *pgxpool.Pool
// MaxPoolConns bounds the pool opened from DSN (ignored when Pool is set,
// and overridden by a pool_max_conns DSN parameter). Zero derives the
// size from the CPU count, clamped to 2–4: publishing is one short
// statement per message, so a few connections suffice.
MaxPoolConns int
// SpillTTL is how long spilled payloads are kept (default 5m).
SpillTTL time.Duration
// JanitorInterval is how often expired spill rows are deleted (default 1m).
JanitorInterval time.Duration
// HealthInterval is how long the listener may sit idle before its
// connection is pinged (default 30s).
HealthInterval time.Duration
// Logger receives operational logs (default slog.Default()).
Logger *slog.Logger
// QueueLen and QueueBytes bound the undelivered messages per local
// subscription (defaults: 10000 messages, 64 MiB).
QueueLen int
QueueBytes int
}
Options configure a Broker.
Source Files