Symfony Doctrine Messenger

https://github.com/symfony/doctrine-messenger
Admin
Provides Doctrine integration for Symfony Messenger, enabling message persistence and retrieval...
Tokens:53,638
Snippets:363
Trust Score:9.5
License:MIT
Update:4 days ago
Show doc for...
Tokens:
Raw
### DoctrineInbox usage example

Source: https://github.com/symfony/doctrine-messenger

Shows how to instantiate DoctrineInbox with a DBAL connection and custom options, then call setup() to create the table if missing.

```php
use Doctrine\DBAL\DriverManager;
use Symfony\Component\Messenger\Bridge\Doctrine\Inbox\DoctrineInbox;

$dbal = DriverManager::getConnection(['url' => 'mysql://user:pass@localhost/app']);

$inbox = new DoctrineInbox(
    $dbal,
    tableName: 'shop_inbox',
    retention: 86400 * 14, // 14 days
    gcProbability: 0.01,   // 1 % chance of GC per record
);
$inbox->setup(); // create the table if missing (TableExistsException swallowed)
```

--------------------------------

### Resolved configuration example for PostgreSQL setup

Source: https://github.com/symfony/doctrine-messenger

Shows how to build a configuration array for a PostgreSQL connection using PostgreSqlConnection::buildConfiguration(). The example includes a DSN with query parameters and an options array. Note that 'use_notify' and 'transport_name' are consumed by the factory before buildConfiguration() is called; passing them directly would trigger an InvalidArgumentException for unknown options.

```php
$config = PostgreSqlConnection::buildConfiguration(
    'doctrine://default?queue_name=emails&redeliver_timeout=1200&auto_setup=1',
    ['check_delayed_interval' => 5000, 'get_notify_timeout' => 10000, 'use_notify' => true, 'transport_name' => 'async'],
);
// equals:
// [
//   'connection' => 'default',
//   'queue_name' => 'emails',
//   'redeliver_timeout' => 1200,
//   'auto_setup' => true,
//   'check_delayed_interval' => 5000,
//   'get_notify_timeout' => 10000,
//   'table_name' => 'messenger_messages',
// ]
```

--------------------------------

### Example DSNs from ConnectionTest

Source: https://github.com/symfony/doctrine-messenger

Example DSNs showing various option combinations, including table_name, redeliver_timeout, queue_name, and auto_setup with boolean values.

```text
doctrine://default
```

```text
doctrine://default?table_name=name_from_dsn&redeliver_timeout=1200&queue_name=normal&auto_setup=false
```

```text
doctrine://default?auto_setup=0
```

```text
doctrine://default?auto_setup=1
```

--------------------------------

### setup() method signature

Source: https://github.com/symfony/doctrine-messenger

Signature of the setup method that builds and creates the table schema.

```php
public function setup(): void
```

--------------------------------

### setup() method implementation

Source: https://github.com/symfony/doctrine-messenger

Builds the table schema via configureSchema() and creates it. Swallows TableExistsException if the table already exists.

```php
$schema = new Schema([], [], $this->connection->createSchemaManager()->createSchemaConfig());
$table = $this->configureSchema($schema, $this->connection, static fn (): bool => true)->getTable($this->tableName);

try {
    $this->connection->createSchemaManager()->createTable($table);
} catch (TableExistsException) {
}
```

--------------------------------

### setup(): void

Source: https://github.com/symfony/doctrine-messenger

Creates the inbox table if it does not exist. Swallows TableExistsException.

```APIDOC
## setup(): void

### Description
Creates the inbox table if it does not exist. Swallows TableExistsException.

### Parameters
None

### Return Value
- `void`
```

--------------------------------

### Usage example for DoctrineSender

Source: https://github.com/symfony/doctrine-messenger

Shows how to instantiate DoctrineSender with a Connection, using the default PhpSerializer.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineSender;

$sender = new DoctrineSender(new Connection($config, $dbal)); // defaults to PhpSerializer
```

--------------------------------

### setup(): void

Source: https://github.com/symfony/doctrine-messenger

Builds the table schema and creates the inbox table if it does not exist. Swallows TableExistsException if the table already exists.

```APIDOC
## setup(): void

### Description
Builds the table schema and creates the inbox table if it does not exist. Swallows TableExistsException if the table already exists.

### Method
setup

### Endpoint
setup(): void

### Parameters
No parameters.

### Response
#### Success Response (200)
- **return** (void) - No return value.

### Request Example
```php
public function setup(): void
```
```

--------------------------------

### setup(): void

Source: https://github.com/symfony/doctrine-messenger

Creates or updates the messenger table (and Oracle sequence) via schema diff. Part of SetupableTransportInterface.

```APIDOC
## setup(): void

### Description
Creates or updates the messenger table (and Oracle sequence) via schema diff. Part of SetupableTransportInterface.

### Method
PHP method

### Signature
```php
public function setup(): void
```

### Return Value
- `void`
```

--------------------------------

### DoctrineReceiver usage example

Source: https://github.com/symfony/doctrine-messenger

Instantiate DoctrineReceiver with a Connection and default PhpSerializer.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineReceiver;

$receiver = new DoctrineReceiver(new Connection($config, $dbal)); // PhpSerializer by default
```

--------------------------------

### Setup, send with DelayStamp, keepalive, sendBatch

Source: https://github.com/symfony/doctrine-messenger

Ensures the table exists with setup(), sends a message with a DelayStamp and retrieves its TransportMessageIdStamp, keeps a long-running handler alive with keepalive() past redeliver_timeout, and sends a batch with sendBatch() returning envelopes stamped with their ids.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;

/** @var DoctrineTransport $transport */
$transport->setup(); // ensure the table exists before sending

$stamped = $transport->send(Envelope::wrap($message, [new DelayStamp(5000)]));
$id = $stamped->last(TransportMessageIdStamp::class)->getId();

// keep a long-running handler alive past redeliver_timeout
$transport->keepalive($stamped, 30);

// batched send returns envelopes each stamped with their id
$sent = $transport->sendBatch([Envelope::wrap($a), Envelope::wrap($b)]);
```

--------------------------------

### buildConfiguration() error handling example

Source: https://github.com/symfony/doctrine-messenger

Catches InvalidArgumentException when an unknown option is provided. The error message lists allowed options.

```php
try {
    Connection::buildConfiguration('doctrine://default', ['new_option' => 'woops']);
} catch (\Symfony\Component\Messenger\Exception\InvalidArgumentException $e) {
    // "Unknown option found: [new_option]. Allowed options are [table_name, queue_name, redeliver_timeout, auto_setup]."
}
```

--------------------------------

### buildConfiguration() usage example

Source: https://github.com/symfony/doctrine-messenger

Builds a configuration array from a DSN. The resulting array contains connection, table_name, redeliver_timeout, queue_name, and auto_setup.

```php
$config = Connection::buildConfiguration(
    'doctrine://default?table_name=name_from_dsn&redeliver_timeout=1200&queue_name=normal&auto_setup=true',
);
// ['connection' => 'default', 'table_name' => 'name_from_dsn', 'redeliver_timeout' => 1200,
//  'queue_name' => 'normal', 'auto_setup' => true]
```

--------------------------------

### Usage example for PostgreSqlNotifyOnIdleListener

Source: https://github.com/symfony/doctrine-messenger

Shows how to instantiate the listener with a logger and null clock. Registration happens automatically via DoctrineTransportFactory, which adds transports created with 'transport_name'.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\EventListener\PostgreSqlNotifyOnIdleListener;

$listener = new PostgreSqlNotifyOnIdleListener($logger, null);
// registration happens automatically via DoctrineTransportFactory:
// $factory = new DoctrineTransportFactory($registry, $listener);
// then transports created with 'transport_name' are added by the factory.
```

--------------------------------

### DoctrineTransportFactory usage example

Source: https://github.com/symfony/doctrine-messenger

Example of instantiating DoctrineTransportFactory with an entity manager as the ConnectionRegistry and a PostgreSqlNotifyOnIdleListener.

```php
use Doctrine\ORM\EntityManagerInterface;
use Symfony\Component\Messenger\Bridge\Doctrine\EventListener\PostgreSqlNotifyOnIdleListener;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransportFactory;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;

$registry = $entityManager; // implements Doctrine\Persistence\ConnectionRegistry
$factory = new DoctrineTransportFactory($registry, new PostgreSqlNotifyOnIdleListener());
```

--------------------------------

### Direct usage of PostgreSqlConnection

Source: https://github.com/symfony/doctrine-messenger

Shows how the factory directly instantiates PostgreSqlConnection with buildConfiguration, then performs a non-blocking drain via get() or blocks up to 30 seconds with waitForNotify().

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\PostgreSqlConnection;

$connection = new PostgreSqlConnection(
    PostgreSqlConnection::buildConfiguration(
        'doctrine://default?check_delayed_interval=5000&get_notify_timeout=10000',
    ),
    $dbal,
);

// non-blocking drain of already-arrived notifications inside the poll
$rows = $connection->get(10);

// or block up to 30 s for a push (delegated to the worker-idle listener in practice)
$wokeUp = $connection->waitForNotify(30000);
var_dump($connection->isListening()); // true after listen()/waitForNotify()
```

--------------------------------

### sendBatch() usage example with TransportMessageIdStamp

Source: https://github.com/symfony/doctrine-messenger

Shows how to call sendBatch() and iterate over the returned envelopes to print ids. Ids are only set on platforms that return them, such as PostgreSQL.

```php
$envelopes = [Envelope::wrap($a), Envelope::wrap($b), Envelope::wrap($c)];
$sent = $sender->sendBatch($envelopes);

foreach ($sent as $envelope) {
    if ($idStamp = $envelope->last(TransportMessageIdStamp::class)) {
        echo $idStamp->getId(), "\n"; // only set on platforms that return ids
    }
}
```

--------------------------------

### send() usage example with DelayStamp and TransportMessageIdStamp

Source: https://github.com/symfony/doctrine-messenger

Shows how to call send() with a DelayStamp and retrieve the resulting TransportMessageIdStamp. The delay is set to 1000 milliseconds.

```php
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;

$sent = $sender->send(Envelope::wrap($message, [new DelayStamp(1000)]));
/** @var TransportMessageIdStamp $idStamp */
$idStamp = $sent->last(TransportMessageIdStamp::class);
printf("queued with id %s\n", $idStamp->getId());
```

--------------------------------

### Usage example with DoctrineReceivedStamp

Source: https://github.com/symfony/doctrine-messenger

Shows how to read the row id from an envelope, compare it with the TransportMessageIdStamp, and manually construct the stamp. Throws LogicException if the stamp is missing.

```php
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineReceivedStamp;
use Symfony\Component\Messenger\Exception\LogicException;
use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;

/** @var Envelope $envelope an envelope received from the Doctrine transport */

// read the row id
$received = $envelope->last(DoctrineReceivedStamp::class);
if (null === $received) {
    throw new LogicException('No DoctrineReceivedStamp found on the Envelope.');
}
echo $received->getId();

// same value as the TransportMessageIdStamp added next to it
$transportId = $envelope->last(TransportMessageIdStamp::class)->getId();
assert($transportId === $received->getId());

// constructing manually (only needed when re-creating the stamp)
$stamp = new DoctrineReceivedStamp('42');
echo $stamp->getId(); // "42"
```

--------------------------------

### get(int $fetchSize = 1): ?array

Source: https://github.com/symfony/doctrine-messenger

Overrides Connection::get() to handle LISTEN/NOTIFY notifications and delayed message checks.

```APIDOC
## get(int $fetchSize = 1): ?array

### Description
Overrides `Connection::get()` to handle LISTEN/NOTIFY notifications and delayed message checks.

### Method
get

### Signature
```php
public function get(int $fetchSize = 1): ?array
```

### Parameters
- **fetchSize** (int) - Optional - Default: 1 - The number of rows to fetch.

### Return Value
- **?array** - The standard doctrine-envelope array or null.
```

--------------------------------

### get() method signature

Source: https://github.com/symfony/doctrine-messenger

Signature for the get() method that fetches envelopes from the Doctrine connection. The fetch size is clamped to at least 1. Async signals are disabled during polling and re-enabled in a finally block. Returns an iterable of Envelope objects, or an empty array when the queue is empty. Throws TransportException on retryable or other DBAL exceptions.

```php
public function get(/* int $fetchSize = 1 */): iterable
```

--------------------------------

### get(int $fetchSize = 1): ?array

Source: https://github.com/symfony/doctrine-messenger

Fetches and marks as delivered up to fetchSize available messages from the queue. Returns an array of doctrine-envelope hashes or null when the queue is empty.

```APIDOC
## get(int $fetchSize = 1): ?array

### Description
Starts a transaction, selects up to `fetchSize` available rows (`queue_name = ?`, `delivered_at IS NULL OR delivered_at < redeliver limit`, `available_at <= now`) ordered by `available_at ASC` with `FOR UPDATE SKIP LOCKED` (falling back to plain `FOR UPDATE`, then to no lock, on platforms that do not support it; `addLockMode()`, line 825). On Oracle the row selection is wrapped in a sub-query to avoid ORA-02014 while keeping the write lock. Each selected row gets `delivered_at = now` in the same transaction, then the transaction commits. Returns the rows as an array of doctrine-envelope hashes (`id`, `body`, `headers` JSON-decoded, `queue_name`, `created_at`, `available_at`, `delivered_at`) or `null` when the queue was empty — in the empty case `queueEmptiedAt` is set (milliseconds since epoch). On `TableNotFoundException` with `auto_setup` enabled, `setup()` runs and the query is retried via `goto get`; otherwise the exception propagates.

### Parameters
- `fetchSize` (int) - Optional (default `1`) - Maximum number of rows to fetch and mark as delivered.

### Returns
- `?array` - Array of doctrine-envelope hashes (`id`, `body`, `headers` JSON-decoded, `queue_name`, `created_at`, `available_at`, `delivered_at`) or `null` when the queue was empty.
```

--------------------------------

### supports method signature

Source: https://github.com/symfony/doctrine-messenger

Signature for checking whether a DSN is supported. Returns true iff the DSN starts with 'doctrine://'.

```php
public function supports(#[\SensitiveParameter] string $dsn, array $options): bool
```

--------------------------------

### get(int $fetchSize = 1): ?array override

Source: https://github.com/symfony/doctrine-messenger

Overrides Connection::get(). When no external listener handles LISTEN/NOTIFY and the queue was last seen empty, registers a LISTEN and drains already-arrived notifications with PDO::getNotify(..., 0) (non-blocking). If no matching notification (channel = table name, payload = queue name) and elapsed time since queue emptied is within check_delayed_interval, waits 1 ms (usleep(1000)) and returns null; otherwise falls through to parent::get(). check_delayed_interval = 0 disables the delayed-check window. Returns the standard doctrine-envelope array or null.

```php
public function get(int $fetchSize = 1): ?array
```

```php
if ($this->notifyHandledExternally || null === $this->queueEmptiedAt) {
    return parent::get($fetchSize);
}

// Fallback: when no external listener handles LISTEN/NOTIFY, only collect the
// notifications that already arrived. ...
$this->executeStatement(\sprintf('LISTEN "%s"', $this->configuration['table_name']));
$this->listening = true;

/** @var \PDO $nativeConnection */
$nativeConnection = $this->driverConnection->getNativeConnection();

$notification = $nativeConnection->getNotify(\PDO::FETCH_ASSOC, 0);
if (
    // no notifications, or for another table or queue
    (false === $notification || $notification['message'] !== $this->configuration['table_name'] || $notification['payload'] !== $this->configuration['queue_name'])
    // delayed messages
    && (microtime(true) * 1000 - $this->queueEmptiedAt < $this->configuration['check_delayed_interval'])
) {
    usleep(1000);
    return null;
}

return parent::get($fetchSize);
```

--------------------------------

### supports(string $dsn, array $options): bool

Source: https://github.com/symfony/doctrine-messenger

Determines whether this factory supports the given DSN. Returns true if the DSN starts with 'doctrine://', which is how Messenger selects the appropriate transport factory.

```APIDOC
## supports(string $dsn, array $options): bool

### Description
Determines whether this factory supports the given DSN. Returns true if the DSN starts with 'doctrine://', which is how Messenger selects the appropriate transport factory.

### Method
supports

### Signature
```php
public function supports(#[
SensitiveParameter] string $dsn, array $options): bool
```

### Parameters
- **dsn** (string) - Required - The DSN to check.
- **options** (array) - Required - Transport options.

### Return Value
- **bool** - `true` iff `str_starts_with($dsn, 'doctrine://')`.
```

--------------------------------

### listen(bool $registerOnDatabase = true): void method

Source: https://github.com/symfony/doctrine-messenger

Registers a LISTEN on the connection for the configured table. Also disables the internal blocking logic of get() (notifyHandledExternally = true), assuming an external listener (e.g. PostgreSqlNotifyOnIdleListener) handles it. Safe to call multiple times; PostgreSQL ignores duplicate LISTEN for the same channel. $registerOnDatabase = false only marks get() as externally handled without executing SQL, "to avoid accumulating unread notifications on connections that will never call waitForNotify()" — exactly how PostgreSqlNotifyOnIdleListener::onWorkerStarted() registers all but the first connection of a multi-transport worker.

```php
public function listen(bool $registerOnDatabase = true): void
```

```php
if ($registerOnDatabase) {
    // This is secure because the table name must be a valid identifier: [...]
    $this->executeStatement(\sprintf('LISTEN "%s"', $this->configuration['table_name']));
    $this->listening = true;
}
$this->notifyHandledExternally = true;
```

--------------------------------

### get(int $fetchSize = 1): iterable

Source: https://github.com/symfony/doctrine-messenger

Fetches a batch of envelopes from the Doctrine transport. Handles retryable exceptions and returns an empty array when the queue is empty.

```APIDOC
## get(int $fetchSize = 1): iterable

### Description
Fetches a batch of envelopes from the Doctrine transport. Handles retryable exceptions and returns an empty array when the queue is empty.

### Method
GET

### Endpoint
`get(int $fetchSize = 1): iterable`

### Parameters
#### Path Parameters
- `$fetchSize` (int) - Optional - The number of messages to fetch, clamped to at least 1.

### Response
#### Success Response (200)
- Returns an iterable of `Envelope` objects, or an empty array when nothing to fetch.

#### Error Response
- Throws `TransportException` on DBAL exceptions or after retry budget exhausted.
```

--------------------------------

### get(int $fetchSize = 1): iterable

Source: https://github.com/symfony/doctrine-messenger

Retrieves up to the specified number of message envelopes from the queue. Returns an iterable of Envelope objects, or an empty iterable when the queue is empty.

```APIDOC
## get(int $fetchSize = 1): iterable

### Description
Retrieves up to the specified number of message envelopes from the queue. Returns an iterable of Envelope objects, or an empty iterable when the queue is empty.

### Method
PHP method

### Signature
```php
public function get(/* int $fetchSize = 1 */): iterable
```

### Parameters
- `$fetchSize` (int) - Optional - Maximum number of envelopes to retrieve. Defaults to 1. Read via `func_get_arg` for backward compatibility.

### Return Value
- `iterable` - An iterable of `Envelope` objects, or an empty iterable when the queue is empty.
```

--------------------------------

### Worker loop with get, ack, reject, getMessageCount, all

Source: https://github.com/symfony/doctrine-messenger

Polls the transport with get(10) and processes envelopes, acknowledging on success and rejecting on exceptions. After the loop, getMessageCount() returns the remaining message count and all() lists envelopes. Requires a DoctrineTransport instance.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;

/** @var DoctrineTransport $transport */
while ($envelopes = $transport->get(10)) {
    foreach ($envelopes as $envelope) {
        try {
            // ... handle $envelope ...
            $transport->ack($envelope);
        } catch (\Throwable $e) {
            $transport->reject($envelope);
        }
    }
}
echo $transport->getMessageCount();       // int
foreach ($transport->all() as $envelope) { /* listing */ }
```

--------------------------------

### Constructor

Source: https://github.com/symfony/doctrine-messenger

Creates a new Connection instance with the given configuration and DBAL connection. The configuration array is merged with DEFAULT_OPTIONS and autoSetup is initialized from the 'auto_setup' key.

```APIDOC
## Constructor

### Description
Creates a new Connection instance with the given configuration and DBAL connection.

### Method
Constructor

### Signature
```php
public function __construct(
    protected array $configuration,
    protected DBALConnection $driverConnection,
)
```

### Parameters
- **configuration** (array) - Required - Options array; merged with `static::DEFAULT_OPTIONS` via `array_replace_recursive()`. Keys: `table_name`, `connection`, `queue_name`, `redeliver_timeout`, `auto_setup`.
- **driverConnection** (doctrine/dbal Connection) - Required - The DBAL connection to the database that hosts the messenger table.

### Usage Example
```php
use Doctrine\DBAL\DriverManager;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;

$dbal = DriverManager::getConnection(['url' => 'mysql://user:pass@localhost/app']);
$connection = new Connection(
    Connection::buildConfiguration('doctrine://default?table_name=messenger_messages'),
    $dbal,
);
```
```

--------------------------------

### Constructor for PostgreSqlNotifyOnIdleListener

Source: https://github.com/symfony/doctrine-messenger

Constructor accepts an optional logger and clock. The logger logs info/warning/debug messages about the wait state; the clock is used for deadline and delayed-message computations, defaulting to microtime(true) when null.

```php
public function __construct(
    private ?LoggerInterface $logger = null,
    private ?ClockInterface $clock = null,
)
```

--------------------------------

### getDriverConnection(): DBALConnection

Source: https://github.com/symfony/doctrine-messenger

Returns the underlying DBAL connection, exposed so the listener can run its own SQL. For example, getEarliestDelayedMessageTime() uses this method.

```APIDOC
## getDriverConnection(): DBALConnection

### Description
Returns the underlying DBAL connection, exposed so the listener can run its own SQL.

### Method
`getDriverConnection()`

### Return Value
- **DBALConnection** - The underlying DBAL connection.

### Example
```php
$connection->getDriverConnection();
```
```

--------------------------------

### buildConfiguration() static method signature

Source: https://github.com/symfony/doctrine-messenger

Static method that parses a DSN and options into a configuration array. Throws InvalidArgumentException on invalid DSN or unknown options.

```php
public static function buildConfiguration(#\[\SensitiveParameter\] string $dsn, array $options = []): array
```

--------------------------------

### Constructor

Source: https://github.com/symfony/doctrine-messenger

Creates a new DoctrineInbox instance with the given DBAL connection and configuration options.

```APIDOC
## Constructor

### Description
Creates a new DoctrineInbox instance with the given DBAL connection and configuration options.

### Parameters
- `connection` (Doctrine\DBAL\Connection) - Required - The DBAL connection that also carries the handlers' transaction.
- `tableName` (string) - Optional - Name of the inbox table. Default: `'messenger_inbox'`.
- `retention` (int) - Optional - Seconds a record is kept; must exceed the time a transport can take to redeliver a message. Default: `604800` (7 days).
- `gcProbability` (float) - Optional - Probability (0 to 1) that a `record()` call also deletes this consumer's expired records. Default: `0.001`.

### Usage Example
```php
use Doctrine\DBAL\DriverManager;
use Symfony\Component\Messenger\Bridge\Doctrine\Inbox\DoctrineInbox;

$dbal = DriverManager::getConnection(['url' => 'mysql://user:pass@localhost/app']);

$inbox = new DoctrineInbox(
    $dbal,
    tableName: 'shop_inbox',
    retention: 86400 * 14, // 14 days
    gcProbability: 0.01,   // 1 % chance of GC per record
);
```
```

--------------------------------

### PostgreSqlConnection::buildConfiguration

Source: https://github.com/symfony/doctrine-messenger

Builds a configuration array for a PostgreSQL connection from a DSN and additional options. This method is part of the public API and is used to resolve the full configuration for the messenger transport.

```APIDOC
## PostgreSqlConnection::buildConfiguration

### Description
Builds a configuration array for a PostgreSQL connection from a DSN and additional options.

### Method
`buildConfiguration` (static)

### Parameters
- **$dsn** (string) - Required - The DSN string, e.g., `'doctrine://default?queue_name=emails&redeliver_timeout=1200&auto_setup=1'`.
- **$options** (array) - Optional - Additional options, e.g., `['check_delayed_interval' => 5000, 'get_notify_timeout' => 10000, 'use_notify' => true, 'transport_name' => 'async']`.

### Return Value
Returns an array with the resolved configuration, including keys like `connection`, `queue_name`, `redeliver_timeout`, `auto_setup`, `check_delayed_interval`, `get_notify_timeout`, `table_name`, etc.

### Request Example
```php
$config = PostgreSqlConnection::buildConfiguration(
    'doctrine://default?queue_name=emails&redeliver_timeout=1200&auto_setup=1',
    ['check_delayed_interval' => 5000, 'get_notify_timeout' => 10000, 'use_notify' => true, 'transport_name' => 'async'],
);
```

### Response Example
```php
// equals:
// [
//   'connection' => 'default',
//   'queue_name' => 'emails',
//   'redeliver_timeout' => 1200,
//   'auto_setup' => true,
//   'check_delayed_interval' => 5000,
//   'get_notify_timeout' => 10000,
//   'table_name' => 'messenger_messages',
// ]
```
```

--------------------------------

### Instantiate Connection with DBAL and DSN

Source: https://github.com/symfony/doctrine-messenger

Shows how to create a Connection instance using DriverManager and buildConfiguration(). The DSN 'doctrine://default?table_name=messenger_messages' selects the 'default' connection name.

```php
use Doctrine\DBAL\DriverManager;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;

$dbal = DriverManager::getConnection(['url' => 'mysql://user:pass@localhost/app']);
$connection = new Connection(
    Connection::buildConfiguration('doctrine://default?table_name=messenger_messages'),
    $dbal,
);
```

--------------------------------

### PostgreSqlNotifyOnIdleListener constructor

Source: https://github.com/symfony/doctrine-messenger

Configures the PostgreSQL notify-on-idle listener with an optional logger and clock. The logger emits informational messages about connection status, and the clock is used for deadline and delayed-message calculations.

```APIDOC
## PostgreSqlNotifyOnIdleListener constructor

### Description
Configures the PostgreSQL notify-on-idle listener with an optional logger and clock. The logger emits informational messages about connection status, and the clock is used for deadline and delayed-message calculations.

### Parameters
- **logger** (`Psr\Log\LoggerInterface`) - Optional (default: `null`) - Optional logger; emits `info` (connection back), `debug` (waiting, with `timeout_ms`), `warning` (connection lost).
- **clock** (`Psr\Clock\ClockInterface`) - Optional (default: `null`) - Optional injectable clock; when `null`, `microtime(true)` is used for deadline/delayed-message math.
```

--------------------------------

### getConfiguration(): array

Source: https://github.com/symfony/doctrine-messenger

Returns the resolved configuration array (defaults merged with the constructor-provided options).

```APIDOC
## getConfiguration(): array

### Description
Returns the resolved configuration array (defaults merged with the constructor-provided options).

### Method
getConfiguration

### Signature
```php
public function getConfiguration(): array
```

### Return Value
array - The resolved configuration array.
```

--------------------------------

### Catch LogicException from onWorkerStarted()

Source: https://github.com/symfony/doctrine-messenger

Shows error handling for the two LogicException conditions thrown by onWorkerStarted() when transports use different DBAL connections or different table_name values. Wrap the worker run in a try/catch to report these configuration errors.

```php
try {
    $worker->run();
} catch (\LogicException $e) {
    // "PostgreSQL transports ... use different DBAL connections ... "
    // "PostgreSQL transports ... use different table_name values ..."
}
```

--------------------------------

### isListening(): bool

Source: https://github.com/symfony/doctrine-messenger

Returns whether a LISTEN is currently registered on the database.

```APIDOC
## isListening(): bool

### Description
Returns whether a `LISTEN` is currently registered on the database (`$this->listening`).

### Method
isListening

### Signature
```php
public function isListening(): bool
```

### Return Value
- **bool** - True if a LISTEN is registered, false otherwise.
```

--------------------------------

### Constructor for DoctrineSender

Source: https://github.com/symfony/doctrine-messenger

Constructor accepting a Connection and optional SerializerInterface, defaulting to PhpSerializer when null.

```php
public function __construct(
    private Connection $connection,
    ?SerializerInterface $serializer = null,
)
```

--------------------------------

### Catch TransportException on receiver->get()

Source: https://github.com/symfony/doctrine-messenger

Wrap receiver polling in a try/catch for TransportException. After 3 consecutive RetryableExceptions or any other DBAL failure, sleep 1 second; the DBAL connection is closed on non-retryable failures so the next poll reconnects.

```php
use Symfony\Component\Messenger\Exception\TransportException;

try {
    $envelopes = $receiver->get(10);
} catch (TransportException $e) {
    // after 3 consecutive RetryableExceptions, or any other DBAL failure
    // (the DBAL connection is closed on non-retryable failures so the next poll reconnects)
    sleep(1);
}
```

--------------------------------

### __construct(array $configuration, DBALConnection $driverConnection)

Source: https://github.com/symfony/doctrine-messenger

Constructor for the PostgreSqlConnection class, inherited from Connection. Initializes the connection with configuration options and a DBAL connection.

```APIDOC
## Constructor

### Description
Initializes a new PostgreSqlConnection instance.

### Method
Constructor

### Signature
```php
public function __construct(
    protected array $configuration,
    protected DBALConnection $driverConnection,
)
```

### Parameters
- **configuration** (array) - Required - Options (may include `check_delayed_interval`, `get_notify_timeout`; merged over `DEFAULT_OPTIONS`).
- **driverConnection** (Doctrine\DBAL\Connection) - Required - The DBAL connection (must be a PostgreSQL connection for LISTEN/NOTIFY to work).
```

--------------------------------

### new PostgreSqlNotifyOnIdleListener(?LoggerInterface $logger = null, ?ClockInterface $clock = null)

Source: https://github.com/symfony/doctrine-messenger

Constructor for the PostgreSqlNotifyOnIdleListener event subscriber. It accepts an optional logger and an optional clock for deadline computations.

```APIDOC
## Constructor

### Description
Creates a new PostgreSqlNotifyOnIdleListener instance.

### Signature
```php
public function __construct(
    private ?LoggerInterface $logger = null,
    private ?ClockInterface $clock = null,
)
```

### Parameters
- **logger** (`Psr\Log\LoggerInterface`) - Optional - Logs info/warning/debug messages about the wait state.
- **clock** (`Psr\Clock\ClockInterface`) - Optional - Injectable clock used for deadline/delayed-message computations; when `null`, `microtime(true)` is used.

### Usage
```php
use Symfony\Component\Messenger\Bridge\Doctrine\EventListener\PostgreSqlNotifyOnIdleListener;

$listener = new PostgreSqlNotifyOnIdleListener($logger, null);
// registration happens automatically via DoctrineTransportFactory:
// $factory = new DoctrineTransportFactory($registry, $listener);
// then transports created with 'transport_name' are added by the factory.
```
```

--------------------------------

### DoctrineInbox constructor

Source: https://github.com/symfony/doctrine-messenger

Configures the Doctrine inbox by providing a DBAL connection, table name, retention period, and garbage collection probability. All parameters are optional except the connection.

```APIDOC
## DoctrineInbox constructor

### Description
Configures the Doctrine inbox by providing a DBAL connection, table name, retention period, and garbage collection probability. All parameters are optional except the connection.

### Parameters
- **connection** (`Doctrine\DBAL\Connection`) - Required - DBAL connection hosting the inbox table; must also carry the handlers' transaction.
- **tableName** (`string`) - Optional (default: `'messenger_inbox'`) - Name of the inbox table.
- **retention** (`int`) - Optional (default: `604800`) - Number of seconds a record is kept; must exceed the time a transport can take to redeliver a message.
- **gcProbability** (`float`) - Optional (default: `0.001`) - Probability, from 0 to 1, that a `record()` call also removes the expired records of its consumer.
```

--------------------------------

### buildConfiguration(string $dsn, array $options = []): array (static)

Source: https://github.com/symfony/doctrine-messenger

Parses a DSN of the form 'doctrine://<connection_name>?<query options>' and merges with extra options and defaults. Throws InvalidArgumentException for invalid DSN or unknown options.

```APIDOC
## buildConfiguration(string $dsn, array $options = []): array (static)

### Description
Parses the DSN with `parse_url()` (throws `InvalidArgumentException` — "The given Doctrine Messenger DSN is invalid." — on failure), parses the query string with `parse_str()`, then builds `['connection' => $params['host']]`, merged with `$query`, `$options`, and `static::DEFAULT_OPTIONS`. `auto_setup` is cast with `filter_var(..., \FILTER_VALIDATE_BOOL)`. Unknown option names — either in `$options` or in the DSN query — throw `InvalidArgumentException` with the message `"Unknown option found: [...] Allowed options are [...]"` and can only be keys of `static::DEFAULT_OPTIONS` (plus `connection`, which is taken from the host).

### Method
buildConfiguration (static)

### Signature
```php
public static function buildConfiguration(#[\SensitiveParameter] string $dsn, array $options = []): array
```

### Parameters
- **dsn** (string) - Required - DSN of the form `doctrine://<connection_name>?<query options>`; the host part selects the Doctrine connection name (`$configuration['connection']`).
- **options** (array) - Optional - Extra options merged over the DSN query string.

### Return Value
array - The configuration array (hash of `table_name`, `connection`, `queue_name`, `redeliver_timeout`, `auto_setup`).

### Request Example
```php
$config = Connection::buildConfiguration(
    'doctrine://default?table_name=name_from_dsn&redeliver_timeout=1200&queue_name=normal&auto_setup=true',
);
// ['connection' => 'default', 'table_name' => 'name_from_dsn', 'redeliver_timeout' => 1200,
//  'queue_name' => 'normal', 'auto_setup' => true]
```

### Error Handling Example
```php
try {
    Connection::buildConfiguration('doctrine://default', ['new_option' => 'woops']);
} catch (\Symfony\Component\Messenger\Exception\InvalidArgumentException $e) {
    // "Unknown option found: [new_option]. Allowed options are [table_name, queue_name, redeliver_timeout, auto_setup]."
}
```
```

--------------------------------

### record() method implementation

Source: https://github.com/symfony/doctrine-messenger

Implementation of record(). Throws LogicException if no transaction is active. Uses a savepoint to avoid aborting the outer transaction on duplicate. Returns false on UniqueConstraintViolationException, otherwise true. Performs conditional GC cleanup.

```php
if (!$this->connection->isTransactionActive()) {
    throw new LogicException(\sprintf('Cannot record message "%s" for "%s" outside of a database transaction. Run the handlers in a transaction on the connection of the inbox, as the "doctrine_transaction" middleware does.', $messageId, $consumer));
}

$now = \DateTimeImmutable::createFromTimestamp(microtime(true));

// the nested transaction is a savepoint, so that a failed insert doesn't abort the transaction of the handlers on PostgreSQL
$this->connection->beginTransaction();
try {
    $this->connection->insert($this->tableName, ['consumer' => $consumer, 'message_id' => $messageId, 'handled_at' => $now], ['handled_at' => Types::DATETIME_IMMUTABLE]);
    $this->connection->commit();
} catch (UniqueConstraintViolationException) {
    $this->connection->rollBack();
    return false;
} catch (\Throwable $e) {
    $this->connection->rollBack();
    throw $e;
}

if ($this->gcProbability > 0 && (1.0 === $this->gcProbability || (random_int(0, \PHP_INT_MAX) / \PHP_INT_MAX) <= $this->gcProbability)) {
    $this->connection->executeStatement("DELETE FROM $this->tableName WHERE consumer = ? AND handled_at < ?", [$consumer, $now->modify(\sprintf('-%d seconds', $this->retention))], [ParameterType::STRING, Types::DATETIME_IMMUTABLE]);
}

return true;
```

--------------------------------

### getConfiguration() method signature

Source: https://github.com/symfony/doctrine-messenger

Returns the resolved configuration array with defaults merged with constructor-provided options.

```php
public function getConfiguration(): array
```

--------------------------------

### PostgreSqlConnection lifecycle operations

Source: https://github.com/symfony/doctrine-messenger

Demonstrates listen(), getDriverConnection(), and reset() lifecycle methods. Serialization always fails with BadMethodCallException.

```php
$connection->listen();                       // SQL LISTEN + mark externally handled
$connection->getDriverConnection();          // access to the raw DBAL connection
$connection->reset();                        // unlisten + clear queue-empty bookkeeping
// serialization and unserialization always fail:
try {
    serialize($connection);
} catch (\BadMethodCallException $e) {
    // "Cannot serialize Symfony\Component\Messenger\Bridge\Doctrine\Transport\PostgreSqlConnection"
}
```

--------------------------------

### listen(bool $registerOnDatabase = true): void

Source: https://github.com/symfony/doctrine-messenger

Registers a LISTEN on the connection for the configured table and disables internal blocking logic.

```APIDOC
## listen(bool $registerOnDatabase = true): void

### Description
Registers a `LISTEN` on the connection for the configured table and disables internal blocking logic of `get()`.

### Method
listen

### Signature
```php
public function listen(bool $registerOnDatabase = true): void
```

### Parameters
- **registerOnDatabase** (bool) - Optional - Default: true - Whether to execute the LISTEN SQL command.
```

--------------------------------

### Constructor

Source: https://github.com/symfony/doctrine-messenger

Creates a new DoctrineTransportFactory instance with a ConnectionRegistry and an optional PostgreSqlNotifyOnIdleListener.

```APIDOC
## Constructor

### Description
Creates a new DoctrineTransportFactory instance.

### Method
Constructor

### Signature
```php
public function __construct(
    private ConnectionRegistry $registry,
    private ?PostgreSqlNotifyOnIdleListener $notifyOnIdleListener = null,
)
```

### Parameters
- **registry** (Doctrine\Persistence\ConnectionRegistry) - Required - Registry whose `getConnection(string $name)` resolves the DSN's host part into a DBAL connection.
- **notifyOnIdleListener** (Symfony\...\Bridge\Doctrine\EventListener\PostgreSqlNotifyOnIdleListener) - Optional - Receives `PostgreSqlConnection`s via `addConnection($transportName, $connection)` so idle workers can block on `LISTEN`/`NOTIFY`.

### Usage Example
```php
use Doctrine\ORM\EntityManagerInterface;
use Symfony\Component\Messenger\Bridge\Doctrine\EventListener\PostgreSqlNotifyOnIdleListener;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransportFactory;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;

$registry = $entityManager; // implements Doctrine\Persistence\ConnectionRegistry
$factory = new DoctrineTransportFactory($registry, new PostgreSqlNotifyOnIdleListener());
```
```

--------------------------------

### Direct construction of DoctrineTransport

Source: https://github.com/symfony/doctrine-messenger

Shows how to instantiate DoctrineTransport directly with a Connection and PhpSerializer. Normally done via DoctrineTransportFactory instead.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;
use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;

$connection = new Connection(
    Connection::buildConfiguration('doctrine://default'),
    $dbalConnection,
);
$transport = new DoctrineTransport($connection, new PhpSerializer());
```

--------------------------------

### DoctrineInbox constructor

Source: https://github.com/symfony/doctrine-messenger

Constructor with parameters for connection, table name, retention, and GC probability. Retention must exceed redelivery time; gcProbability is between 0 and 1.

```php
/**
 * @param int   $retention     The number of seconds a record is kept, which must exceed the time a transport can take to redeliver a message
 * @param float $gcProbability The probability, from 0 to 1, that recording a message removes the expired records of its consumer
 */
public function __construct(
    private Connection $connection,
    private string $tableName = 'messenger_inbox',
    private int $retention = 604800,
    private float $gcProbability = 0.001,
)
```

--------------------------------

### createTransport implementation logic

Source: https://github.com/symfony/doctrine-messenger

Builds the Doctrine transport. Reads use_notify (default true) and transport_name (default null) from options, unsets them, then builds configuration via PostgreSqlConnection::buildConfiguration(). Falls back to a standard Connection when use_notify is false or the platform is not PostgreSQL. Connection resolution failures are rethrown as TransportException.

```php
$useNotify = $options['use_notify'] ?? true;
$transportName = $options['transport_name'] ?? null;
unset($options['transport_name'], $options['use_notify']);
// Always allow PostgreSQL-specific keys, to be able to transparently fallback to the native driver when LISTEN/NOTIFY isn't available
$configuration = PostgreSqlConnection::buildConfiguration($dsn, $options);

try {
    $driverConnection = $this->registry->getConnection($configuration['connection']);
} catch (\InvalidArgumentException $e) {
    throw new TransportException('Could not find Doctrine connection from Messenger DSN.', 0, $e);
}

if ($useNotify && $driverConnection->getDatabasePlatform() instanceof PostgreSQLPlatform) {
    $connection = new PostgreSqlConnection($configuration, $driverConnection);
    if (null !== $transportName) {
        $this->notifyOnIdleListener?->addConnection($transportName, $connection);
    }
} else {
    $connection = new Connection($configuration, $driverConnection);
}

return new DoctrineTransport($connection, $serializer);
```

--------------------------------

### isListening(): bool method

Source: https://github.com/symfony/doctrine-messenger

Returns whether a LISTEN is currently registered on the database ($this->listening). This is what PostgreSqlNotifyOnIdleListenerTest asserts after listen().

```php
public function isListening(): bool
```

--------------------------------

### reset() method signature

Source: https://github.com/symfony/doctrine-messenger

Clears $this->queueEmptiedAt to null. Called by the service container between requests; no return value.

```php
public function reset(): void
```

--------------------------------

### Basic creation with doctrine:// DSN

Source: https://github.com/symfony/doctrine-messenger

Creates a DoctrineTransport using a plain DSN. Returns a DoctrineTransport over a plain Connection, or PostgreSqlConnection when the database is PostgreSQL and use_notify defaults to true.

```php
$transport = $factory->createTransport('doctrine://default', [], $serializer);
// -> DoctrineTransport over a plain Connection (or PostgreSqlConnection when the DB is PostgreSQL and use_notify defaults to true)
```

--------------------------------

### Wire DoctrineInbox with Messenger middleware stack

Source: https://github.com/symfony/doctrine-messenger

Sets up a MessageBus with SendMessageMiddleware, InboxMiddleware, and HandleMessageMiddleware. Requires an active database transaction on the same connection as the inbox; the comment notes that handlers must run in that transaction.

```php
use Symfony\Component\Messenger\Bridge\Doctrine\Inbox\DoctrineInbox;
use Symfony\Component\Messenger\Handler\HandlersLocator;
use Symfony\Component\Messenger\MessageBus;
use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;
use Symfony\Component\Messenger\Middleware\InboxMiddleware;
use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Transport\InMemoryTransport;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;

$inbox = new DoctrineInbox($dbal); // must run handlers in the same transaction/connection
$inbox->setup();

$bus = new MessageBus([
    new SendMessageMiddleware(new InMemoryTransportFactory(), true), // actual transport enclosed
    new InboxMiddleware($inbox), // deduplicates by (consumer, messageId)
    new HandleMessageMiddleware(new HandlersLocator([
        SendEmailMessage::class => $handler,
    ])),
]);
```

--------------------------------

### Serialization guards: __serialize and __unserialize

Source: https://github.com/symfony/doctrine-messenger

Both methods always throw \BadMethodCallException('Cannot serialize '.__CLASS__). PHP object serialization of this class is forbidden.

```php
public function __serialize(): array
public function __unserialize(array $data): void
```
Tokens:
Raw