ergon_pgmq_sup (ergon v0.5.0)
View SourceEverything one call to ergon:start_consumer/2 creates.
ergon_pgmq_sup (rest_for_one)
|- pgo pool only when long polling; a single dedicated connection
|- wpool concurrency x ergon_pgmq_runner
\- consumer the batch cyclerest_for_one in dependency order, as in ergon_queue_sup: the consumer casts
into the pool, so the pool must exist first and the consumer must come back with
it. A consumer crash leaves the runners to finish what they hold.
The long-poll connection
A long-polling read blocks server-side for up to max_poll_seconds, holding its
connection for the whole call. Issued against the shared pool that would take a
connection out of circulation for every other query on the node, and with a few
such consumers it would empty the pool entirely.
So a long-polling consumer gets a pgo pool of its own, sized one, and its reads
are pinned to it through the query_options() that ergon_sql:query/3 threads
down to the driver. Consumers that do not long poll share the main pool as
everything else does: their reads return immediately.
pgo pools live under pgo_sup rather than here, for the same reason
ergon_repo's does, so this supervisor starts it in init/1 and does not list
it as a child.
Summary
Functions
The registered name of a consumer's executor pool.
The registered name of a long-polling consumer's dedicated read pool.
Types
-type pg_null() :: null.
-type pgmq_handler() :: fun((pgmq_message()) -> ok | {error, binary()}).
-type pgmq_message() :: #{id := non_neg_integer(), read_ct := non_neg_integer(), message := json:decode_value(), headers := json:decode_value() | pg_null()}.
-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 init({pgmq_queue(), pgmq_handler()}) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
The registered name of a consumer's executor pool.
The registered name of a long-polling consumer's dedicated read pool.
-spec start_link(pgmq_queue(), pgmq_handler()) -> {ok, pid()} | {error, term()}.