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

A `GenStage` producer for Broadway that consumes messages from PgQue.

## Options

* `:repo` - The Ecto.Repo module or Postgrex connection to use for database operations.

* `:connection_opts` (`t:keyword/0`) - Optional keyword list of Postgrex connection options to start a dedicated connection pool for this producer.

* `: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"`.

* `: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 `fn(repo, batch_id, msg_id) -> term` 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`.

* `:transformer` - Callback `{module, function, opts}` or function 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*