ergon_pgmq_consumer_sup (ergon v0.5.0)

The dynamic supervisor pgmq consumers run under.

The pgmq sibling of ergon_worker_sup, and for the same reason: which queues to consume, with what concurrency and which handler, is the host's decision rather than the library's, so consumers are added at runtime instead of listed here.

Each call adds one ergon_pgmq_sup (an executor pool and its consumer, plus a dedicated read pool when long polling) supervised independently, so one queue collapsing does not disturb the others.

transient restart: a consumer tree that exits abnormally comes back, one shut down deliberately stays down.

Summary

Types

pg_null()

pgmq_handler()

pgmq_message()

pgmq_queue()

pgmq_read_strategy()

Functions

init/1

start_consumer/2

Start a supervised consumer draining Queue, running Handler on each message.

start_link()

stop_consumer(Pid)

Stop a consumer started by start_consumer/2, along with its pools.

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([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.

start_consumer/2

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

Start a supervised consumer draining Queue, running Handler on each message.

Returns the pid of the consumer's supervisor, which is what stop_consumer/1 takes.

start_link()

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

stop_consumer(Pid)

-spec stop_consumer(pid()) -> ok | {error, not_found}.

Stop a consumer started by start_consumer/2, along with its pools.