Lightweight Postgres client executing PgQue queue operations for Broadway.
Supports shared Ecto repositories, direct Postgrex connection PIDs, and dedicated connection pools.
Summary
Functions
Acknowledges and closes an active batch.
Ensures a PgQue queue exists and configures its ticker responsiveness.
Moves an event directly to the Dead-Letter Queue (DLQ).
Marks an event for retry after a delay in seconds.
Executes an arbitrary SQL query against an Ecto repo or Postgrex connection.
Receives a batch of messages from a queue for the given consumer.
Receives a batch of messages cooperatively as a subconsumer within a consumer group.
Sets a configuration parameter for a queue.
Registers a consumer subscription for a queue.
Types
Functions
@spec ack(repo_or_conn(), integer(), keyword()) :: {:ok, integer()} | {:error, term()}
Acknowledges and closes an active batch.
@spec create_queue(repo_or_conn(), queue_name(), keyword()) :: {:ok, term()} | {:error, term()}
Ensures a PgQue queue exists and configures its ticker responsiveness.
@spec event_dead(repo_or_conn(), integer(), integer(), String.t(), keyword()) :: {:ok, integer()} | {:error, term()}
Moves an event directly to the Dead-Letter Queue (DLQ).
@spec event_retry(repo_or_conn(), integer(), integer(), integer(), keyword()) :: {:ok, integer()} | {:error, term()}
Marks an event for retry after a delay in seconds.
@spec query(repo_or_conn(), String.t(), [term()], keyword()) :: {:ok, Postgrex.Result.t()} | {:error, term()}
Executes an arbitrary SQL query against an Ecto repo or Postgrex connection.
@spec receive(repo_or_conn(), queue_name(), consumer_name(), pos_integer(), keyword()) :: {:ok, {integer() | nil, [map()]}} | {:error, term()}
Receives a batch of messages from a queue for the given consumer.
@spec receive_coop( repo_or_conn(), queue_name(), consumer_name(), String.t(), pos_integer(), keyword() ) :: {:ok, {integer() | nil, [map()]}} | {:error, term()}
Receives a batch of messages cooperatively as a subconsumer within a consumer group.
@spec set_queue_config( repo_or_conn(), queue_name(), String.t(), String.t(), keyword() ) :: {:ok, term()} | {:error, term()}
Sets a configuration parameter for a queue.
@spec subscribe(repo_or_conn(), queue_name(), consumer_name(), keyword()) :: {:ok, integer()} | {:error, term()}
Registers a consumer subscription for a queue.