Documentation
¶
Overview ¶
Package pgchannel — see subject_match.go for the package doc.
This file is the providers.RuntimeBroker implementation. Connect opens a pgxpool against ClusterConfig.dsn; EnsureStream + EnsureConsumer upsert the metadata tables; Publish INSERTs into eventbus_events + NOTIFYs the per-stream channel; Subscribe acquires a per-consumer advisory lock, spawns a LISTEN goroutine, and runs a polling loop that delivers events through the MessageHandler with at-least-once semantics + max_deliver enforcement; Ack parses the "<stream>:<consumer>:<id>" token and advances the consumer cursor scoped to that stream.
Package pgchannel — providers.RuntimeBroker implementation backed by Postgres LISTEN/NOTIFY + polling fallback + advisory locks.
This file holds the subject-matcher: converts the ConsumerConfig.filter_subject DSL into a parameterised SQL predicate applied inside the polling SELECT.
Supported filter syntax (intentionally minimal for v0.2.0):
- literal: "bmw.fulfillment.created" → subject = $N
- prefix wildcard: "bmw.>" → subject LIKE 'bmw.%'
- comma-separated: "bmw.delivered,bmw.cancelled" → subject = ANY($N::text[])
Single-token wildcard (`bmw.*`) is REJECTED by GenFilterValidated; the pgchannel provider does not implement per-segment matching in v0.2.0. Use prefix-wildcard or comma-list instead.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func GenFilter ¶
GenFilter generates a parameterised SQL predicate fragment + arg slice for the given filter_subject. The caller supplies the placeholder string (e.g. "$1") that the returned SQL will reference; the caller is responsible for stitching args into the surrounding query.
The returned sql is always wrapped in parentheses so it can be AND-ed into a larger WHERE clause without ambiguity.
GenFilter does NOT validate against the single-token-wildcard "*" form; for input that may originate from user config use GenFilterValidated.
func GenFilterValidated ¶
GenFilterValidated wraps GenFilter with strict validation: rejects the single-token wildcard form ("bmw.*") with an error that points users to the supported alternatives.
Callers that receive filter_subject from user config should call this variant; internal callers that have already validated their input may call GenFilter directly.
func NewRuntime ¶
func NewRuntime() providers.RuntimeBroker
NewRuntime returns a fresh providers.RuntimeBroker backed by Postgres LISTEN/NOTIFY + a polling fallback + advisory locks. Each Connect call opens a new pgxpool — runtimes themselves are stateless.
Types ¶
type Connection ¶
type Connection struct {
// contains filtered or unexported fields
}
Connection is the pgchannel-specific providers.Connection. It wraps a pgxpool.Pool plus the DSN + poll cadence configured at Connect time, and is passed back through every RuntimeBroker call as the opaque handle.
Callers in the providers/pgchannel package may use the exported Pool() getter to access the underlying pool for SQL operations. Cross-package callers should only interact via the providers.RuntimeBroker interface.
func OpenConnection ¶
OpenConnection opens a new pgxpool.Pool against dsn and returns a pgchannel.Connection. maxConns caps the pool size; pass 0 (or any non-positive value) to accept the package default (defaultMaxConns, 16 — sized so a handful of Subscribe consumers each holding 2 connections (advisory lock + LISTEN) plus Publish/Ack/Ensure transients fit without pool exhaustion).
The returned Connection is safe for concurrent use; Close releases the underlying pool.
func (*Connection) Close ¶
func (c *Connection) Close() error
Close releases the underlying pgxpool.Pool. Idempotent — pgxpool's own Close handles repeat calls safely.
func (*Connection) DSN ¶
func (c *Connection) DSN() string
DSN returns the DSN this connection was opened with. Used by Subscribe when it needs a dedicated LISTEN connection outside the shared pool.
func (*Connection) PollInterval ¶
func (c *Connection) PollInterval() time.Duration
PollInterval returns the per-consumer polling cadence configured at Connect time. Used by the Subscribe loop in runtime.go.
func (*Connection) Pool ¶
func (c *Connection) Pool() *pgxpool.Pool
Pool exposes the underlying pgxpool.Pool for use by sibling files in the providers/pgchannel package (polling.go, runtime.go). Cross-package callers should NOT use this; route through providers.RuntimeBroker.
func (*Connection) Provider ¶
func (c *Connection) Provider() string
Provider returns the static provider identifier "pgchannel".
func (*Connection) SetPollInterval ¶
func (c *Connection) SetPollInterval(d time.Duration)
SetPollInterval overrides the poll cadence after construction. Used by runtime.Connect to thread the ClusterConfig.poll_interval value through.
type EventRow ¶
type EventRow struct {
ID int64
StreamName string
Subject string
Headers json.RawMessage
Payload []byte
CorrelationID string
Ts time.Time
}
EventRow is a single row from eventbus_events as read by pollOnce. Headers are kept as raw JSON so the runtime can decide whether to decode them (publish path always re-encodes; subscribe path decodes once per delivery before invoking the handler).
Source Files
Directories