ergon_pgmq_sup (ergon v0.5.0)

View Source

Everything 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 cycle

rest_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

pg_null()

-type pg_null() :: null.

pgmq_handler()

-type pgmq_handler() :: fun((pgmq_message()) -> ok | {error, binary()}).

pgmq_message()

-type pgmq_message() ::
          #{id := non_neg_integer(),
            read_ct := non_neg_integer(),
            message := json:decode_value(),
            headers := json:decode_value() | pg_null()}.

pgmq_queue()

-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()}.

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

init/1

-spec init({pgmq_queue(), pgmq_handler()}) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.

pool_name(QueueName)

-spec pool_name(binary()) -> atom().

The registered name of a consumer's executor pool.

read_pool_name(QueueName)

-spec read_pool_name(binary()) -> atom().

The registered name of a long-polling consumer's dedicated read pool.

start_link(Queue, Handler)

-spec start_link(pgmq_queue(), pgmq_handler()) -> {ok, pid()} | {error, term()}.