# `BroadwayPgQue.Producer`
[🔗](https://github.com/kuku/broadway_pgque/blob/v0.1.1/lib/broadway_pgque/producer.ex#L1)

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` (`t: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` (`t:pos_integer/0`) - Base polling interval in milliseconds when idle. The default value is `1000`.

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

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

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

* `:ticker_max_lag` (`t: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` (`t:timeout/0`) - Query timeout in milliseconds for PgQue SQL operations. The default value is `15000`.

* `:acknowledger` (`t: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` (`t: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.

# `transform`

Default transformer function that wraps a raw PgQue message map into a `Broadway.Message`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*