A GenStage producer for Broadway that consumes messages from PgQue.

Options

• :pool - Module implementing BroadwayPgQue.Pool (e.g. BroadwayPgQue.Pool.Postgrex, BroadwayPgQue.Pool.Ecto, or custom). The default value is nil.

• :pool_opts (keyword/0) - Keyword list of options passed to the pool module's init/1 callback. The default value is [].

• :repo - The Ecto.Repo module to use for database operations. The default value is nil.

• :conn - Direct Postgrex connection process PID or registered name (no Ecto required). The default value is nil.

• :connection_opts - Optional keyword list of Postgrex connection options to start a dedicated connection pool for this producer. The default value is nil.

• :pool_type - Connection pooling strategy: :shared_repo (uses Ecto repo pool), :dedicated (spawns an isolated Postgrex worker), or :conn (direct PID). The default value is :shared_repo.

• :queue - Required. Name of the PgQue queue.

• :consumer - Name of the consumer registration. Defaults to "broadway_<queue>".

• :subconsumer - Name of the subconsumer for cooperative consumer groups (pgque.receive_coop). The default value is nil.

• :receive_mode - Receive mode: :auto (uses :coop if :subconsumer is provided, :standard otherwise), :standard (pgque.receive), or :coop (pgque.receive_coop). The default value is :auto.

• :poll_interval (pos_integer/0) - Base polling interval in milliseconds when idle. The default value is 1000.

• :max_poll_interval (pos_integer/0) - Maximum backoff interval in milliseconds. The default value is 30000.

• :backoff_factor (float/0) - Exponential backoff multiplier applied when consecutive polls yield no messages. The default value is 1.5.

• :max_batch_size (pos_integer/0) - Maximum number of messages fetched per poll. The default value is 100.

• :ticker_max_lag (String.t/0) - Maximum lag interval for PgQue ticker responsiveness. The default value is "0.1 seconds".

• :history_tables - Number of partition tables in the queue ring-buffer (alias for :ntables, minimum 3). The default value is nil.

• :ntables - Number of partition tables in the queue ring-buffer (default 3). The default value is nil.

• :rotation_period - Interval for ring-buffer table rotation (e.g. "2 hours", "1 day"). The default value is nil.

• :query_timeout (timeout/0) - Query timeout in milliseconds for PgQue SQL operations. The default value is 15000.

• :acknowledger (atom/0) - Broadway acknowledger module. The default value is BroadwayPgQue.Acknowledger.

• :on_failure - Default action for unhandled failed messages: :dead (moves to DLQ), :retry (schedules retry with :default_retry_delay), or a custom callback. The default value is :dead.

• :default_retry_delay (pos_integer/0) - Default retry delay in seconds when retrying failed messages. The default value is 60.

• :poll_type - Polling strategy when the queue is empty: :fixed (keeps constant poll_interval) or :adaptive (exponentially backs off via backoff_factor up to max_poll_interval). The default value is :fixed.

• :transformer - Callback to transform raw PgQue message maps into Broadway.Message structs. The default value is {BroadwayPgQue.Producer, :transform, []}.

Instant Wake Up

When producers are in idle polling backoff, sending :wake_up to the producer process will immediately reset its backoff timer and poll for messages.