ergon_pgmq_queue (ergon v0.5.0)
View SourceRuntime configuration for one pgmq consumer. Mirrors ergon_queue, so both
kinds of queue are configured the same way.
Q = ergon_pgmq_queue:with_concurrency(
ergon_pgmq_queue:with_notify(ergon_pgmq_queue:new(~"events")), 4),
{ok, _} = ergon:start_consumer(Q, fun handle/1).Choosing a wake path
Three ways to find out a message has arrived, in increasing order of what they cost the database:
- Poll only (the default).
poll_intervalbetween reads, 100 ms out of the box. Latency is the interval; the price is one query per interval per consumer even when the queue is empty. - Notify (
with_notify/1,2). Ergon's singleergon.notify_pending_pgmq()tick wakes the consumer, sopoll_intervalcan be raised to seconds without losing responsiveness. Latency floor is the tick's 1 s, which is pg_cron's finest granularity. One sharedLISTENconnection serves every consumer on the node. - Long poll (
with_long_poll/1,2,3). The read itself blocks server-side until a message arrives. Lowest latency (100 ms by default) and no repeated round-trips, but the connection is held for the whole call, so the consumer gets a single-connection pool of its own.
Long poll is the better choice for a handful of latency-sensitive queues; notify is the better choice for many mostly-idle ones, since it costs one connection per node rather than one per consumer.
Summary
Functions
The handler deadline actually applied: the configured one, or the visibility timeout when none was set.
Whether this consumer blocks server-side on read, and so needs its own pool.
A consumer for Name: polled every 100 ms, ten messages per read, a 30 s
visibility timeout, one handler at a time, and no wake path beyond the poll.
Set how many messages one read takes.
Set how many handlers may run at once. Sizes the executor pool.
Set a per-message deadline in milliseconds, after which the handler is killed.
Block server-side on each read until a message arrives, for up to 5 seconds, checking every 100 ms.
Like with_long_poll/1 with an explicit block, in seconds.
Like with_long_poll/2 with an explicit server-side check interval, in
milliseconds. That interval is the latency floor.
Wake this consumer through Ergon's notification tick, on the default channel
pgmq_<queue>.
Like with_notify/1 on an explicit channel.
Set how long the consumer waits between reads, in milliseconds.
Set the read strategy, i.e. whether and how messages are ordered.
Set how long a read message stays hidden, in seconds.
Types
-type pgmq_queue() :: #{name := binary(), poll_interval := pos_integer(), batch_size := pos_integer(), visibility_timeout := pos_integer(), concurrency := pos_integer(), handler_timeout := timeout() | undefined, notify_channel := binary() | undefined, read_strategy := pgmq_read_strategy()}.
-type pgmq_read_strategy() :: plain | grouped | grouped_head | grouped_rr | {long_poll, MaxSeconds :: pos_integer(), IntervalMs :: pos_integer()} | {long_poll, plain | grouped | grouped_head | grouped_rr, pos_integer(), pos_integer()}.
Functions
-spec effective_handler_timeout(pgmq_queue()) -> timeout().
The handler deadline actually applied: the configured one, or the visibility timeout when none was set.
-spec long_poll(pgmq_queue()) -> boolean().
Whether this consumer blocks server-side on read, and so needs its own pool.
-spec new(binary()) -> pgmq_queue().
A consumer for Name: polled every 100 ms, ten messages per read, a 30 s
visibility timeout, one handler at a time, and no wake path beyond the poll.
handler_timeout is left undefined here and resolves to the visibility
timeout unless set explicitly. See effective_handler_timeout/1.
-spec with_batch_size(pgmq_queue(), pos_integer()) -> pgmq_queue().
Set how many messages one read takes.
Also the width of a batch cycle: the consumer dispatches the whole batch, waits
for all of it, and archives the successes in one call. Raising it amortises
round-trips; it does not raise concurrency, which is with_concurrency/2.
-spec with_concurrency(pgmq_queue(), pos_integer()) -> pgmq_queue().
Set how many handlers may run at once. Sizes the executor pool.
-spec with_handler_timeout(pgmq_queue(), timeout()) -> pgmq_queue().
Set a per-message deadline in milliseconds, after which the handler is killed.
Rarely needed: the default is the visibility timeout, which is already the right answer. Past that point pgmq has made the message visible again and another consumer may hold it, so continuing is pointless and archiving afterwards would remove a message someone else is still working.
-spec with_long_poll(pgmq_queue()) -> pgmq_queue().
Block server-side on each read until a message arrives, for up to 5 seconds, checking every 100 ms.
Gives the consumer a single-connection pool of its own, because the call holds
its connection for the whole block. Do not combine with with_notify/1: the read
already returns the instant work appears, so a notification has nothing to add.
-spec with_long_poll(pgmq_queue(), pos_integer()) -> pgmq_queue().
Like with_long_poll/1 with an explicit block, in seconds.
-spec with_long_poll(pgmq_queue(), pos_integer(), pos_integer()) -> pgmq_queue().
Like with_long_poll/2 with an explicit server-side check interval, in
milliseconds. That interval is the latency floor.
Preserves any grouped strategy already set, so ordering and long polling compose.
-spec with_notify(pgmq_queue()) -> pgmq_queue().
Wake this consumer through Ergon's notification tick, on the default channel
pgmq_<queue>.
Register the queue with ergon_pgmq:enable_notify/1 as well; this side only
subscribes. Pair it with a longer with_poll_interval/2, because the point is to stop
polling an idle queue, not to poll it at 100 ms and listen.
-spec with_notify(pgmq_queue(), binary()) -> pgmq_queue().
Like with_notify/1 on an explicit channel.
-spec with_poll_interval(pgmq_queue(), pos_integer()) -> pgmq_queue().
Set how long the consumer waits between reads, in milliseconds.
-spec with_read_strategy(pgmq_queue(), plain | grouped | grouped_head | grouped_rr) -> pgmq_queue().
Set the read strategy, i.e. whether and how messages are ordered.
plain takes whatever is visible. grouped_head, grouped_rr and grouped
order strictly within each x-pgmq-group while letting different groups run in
parallel. See pgmq_read_strategy(). Send with ergon_pgmq:group_header/1 to
put a message in a group.
Preserves any long-poll setting, so the two compose in either order.
-spec with_visibility_timeout(pgmq_queue(), pos_integer()) -> pgmq_queue().
Set how long a read message stays hidden, in seconds.
This is the redelivery clock and therefore the failure-handling mechanism: a message not archived within it becomes visible again and is delivered to someone else. It should comfortably exceed how long a handler takes, or work will be duplicated rather than retried.