Documentation
¶
Overview ¶
Package consume runs a JetStream pull consumer with packtrail's lifecycle: optional pinned priority group (one active consumer per partition with managed failover), bounded concurrency, panic isolation and a graceful drain on shutdown — in-flight handlers keep running under a context detached from the shutdown signal and are only cancelled when the drain budget is exhausted (I-15).
Index ¶
Constants ¶
const ( // DefaultPullExpiry is the default lifetime of a pull request. DefaultPullExpiry = 5 * time.Second // DefaultAckWait is the default ack wait. Handlers send InProgress while // they run, so the ack wait only bounds how long a message held by a dead // process waits before redelivery — not how long a handler may take. DefaultAckWait = 5 * time.Second )
const DefaultDrain = 30 * time.Second
DefaultDrain is the default drain budget.
Variables ¶
This section is empty.
Functions ¶
func Run ¶
Run creates (or updates) the consumer and processes messages until ctx is done, then drains.
func ShuttingDown ¶
ShuttingDown returns a channel closed when the runner that called the handler starts shutting down. Handler contexts are detached from shutdown (in-flight work drains); a handler that is only *waiting* (not working) uses this to give its message back at once (F2-06).
Types ¶
type Config ¶
type Config struct {
Stream string
Consumer jetstream.ConsumerConfig
// Group, when set, makes the consumer a pinned priority group: only one
// client receives messages at a time, another takes over when it stops
// pulling.
Group string
// Concurrency bounds parallel handlers (<= 1: strictly serial).
Concurrency int
Drain time.Duration
// PullExpiry bounds how long a pull request lives on the server: a
// request left behind by a stopped client can absorb a message until the
// ack wait expires (0 = DefaultPullExpiry).
PullExpiry time.Duration
// KeepExisting uses an existing consumer as configured instead of
// updating it (consumers shared by many processes, e.g. workers).
KeepExisting bool
// OnReady receives the consumer's effective configuration.
OnReady func(*jetstream.ConsumerInfo)
// Pulling is called once, when the consumer starts pulling messages.
Pulling func()
Logger *slog.Logger
Handler Handler
}
Config configures Run.
Source Files