Documentation
¶
Overview ¶
Package pgx provides a distributed stream.Service backed by Postgres.
The caller owns the *pgxpool.Pool; this package never opens or closes the connection and never reads DATABASE_URL.
Claim uses FOR UPDATE SKIP LOCKED so multiple workers (in the same process or across processes) can compete safely on the same consumer group without double-delivery. LISTEN/NOTIFY wakes subscribers immediately on Publish; PollInterval is the fallback wake-up cadence for missed notifications and stuck-pending reclaim.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrClosed = errors.New("stream/pgx: service closed")
ErrClosed is returned by Publish or Subscribe after Close.
var Schema string
Functions ¶
func Apply ¶
Apply executes the embedded schema against pool. Idempotent.
Boot-fast path: a single to_regclass('stream_messages') probe checks whether the canonical table already exists. The CREATE statements inside Schema are all IF NOT EXISTS, but each still walks pg_class and takes a brief share lock — adding up to a measurable hit on every process start when nothing has actually changed.
Schema-monotonicity assumption: if stream_messages is present we skip the entire schema, so new indexes/columns added in a later release will not be applied to existing installs by this function. When the schema evolves, ship the change through a separately named migration (e.g. ApplyV2) and invoke both at startup.
Types ¶
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is the pgx stream.Service implementation.
func New ¶
func New(pool *pgxpool.Pool, opts ...stream.ServiceOption) *Service
New constructs a stream/pgx Service. The caller owns pool.
func (*Service) Close ¶
Close shuts down consumers, stops the listener, and releases the listener's pool connection. Idempotent. ctx bounds the consumer-drain phase: cancellation aborts before the next consumer is drained and returns ctx.Err(). The listener is always stopped before returning so the pool connection isn't leaked.
func (*Service) Publish ¶
func (s *Service) Publish(ctx context.Context, name string, payload []byte, opts ...stream.PublishOption) error
Publish appends a message and emits a NOTIFY for cross-process wake-ups, then signals local subscribers and runs trim if MaxLen/ MaxAge are set.
func (*Service) Subscribe ¶
func (s *Service) Subscribe(ctx context.Context, name, group string, h stream.Handler, opts ...stream.SubscribeOption) (stream.Consumer, error)
Subscribe lazily creates the consumer group at the requested start position, ensures the LISTEN connection is up, and spawns concurrency-many workers.
Source Files