README

Symfony Messenger AMQP transport that uses consume instead of get.

Overview

the-consoomer is a custom AMQP transport for the Symfony Messenger component. Unlike the default AMQP transport, which relies on the get method for message retrieval, this package uses the consume method to process messages from an AMQP broker. This can result in different performance characteristics and is more suitable for certain messaging patterns.

Requirements: PHP 8.4+ with the amqp extension (PECL amqp) ^2.0 installed.

• Language: PHP
• Framework: Symfony
• License: MIT
• Status: Public, actively maintained

Features

• Custom AMQP transport for Symfony Messenger
• Uses basic_consume for push-based message processing
• Lower latency and better throughput than polling-based basic_get

Installation

composer require crazy-goat/the-consoomer

Usage

1. Register the transport factory in your Symfony services configuration.
2. Use it in your Messenger transport configuration.

Service registration (config/services.yaml):

services:
    CrazyGoat\TheConsoomer\AmqpTransportFactory:
        tags:
            - { name: 'messenger.transport_factory' }

Messenger configuration (config/packages/messenger.yaml):

framework:
    messenger:
        transports:
            consoomer:
                dsn: 'amqp-consoomer://guest:guest@localhost:5672/%2f/messages?queue=my_queue'

DSN format

amqp-consoomer://<user>:<password>@<host>:<port>/<vhost>/<exchange>/?queue=<queue_name>

Example: amqp-consoomer://guest:guest@localhost:5672/%2f/my_exchange/?queue=test

The path has at most two segments, <vhost>/<exchange> (the trailing slash is optional). A leading // means the default vhost / (e.g. ...//my_exchange). Any additional segment is a configuration error and throws InvalidArgumentException instead of being silently discarded (#234).

Reserved query keys: the query string must not use the keys host, port, user, password, vhost or exchange — these come from the DSN authority/path and are authoritative. A query parameter with one of these keys (e.g. ?host=evil&password=secret) throws InvalidArgumentException at parse time instead of being applied (#207/#360) — a previously-silent connection/credential override was a security hole, so this is an intentional BC break rather than a deprecation.

Options

Every key below is accepted both in the DSN query string and programmatically (the $options argument of AmqpTransportFactory::create()); allow_insecure_verify is programmatic-only by design. Options are grouped by area; the retry* keys have their own table in Retry Configuration and the ssl* keys in SSL/TLS.

Connection

The factory does not connect at construction: the first operation that needs the broker establishes the connection lazily, wrapped in the configured retry, so a briefly-unavailable broker does not fail app boot (#230).

Topology

Consuming

Publishing

Routing Key Resolution

The transport uses separate routing keys for consuming and sending:

• Consumer (Receiver): Uses routing_key as the binding key when declaring/binding the queue to an exchange. This determines which messages are routed to the queue.
• Sender: Uses default_publish_routing_key as the default routing key when publishing messages. This determines how messages are routed through the exchange.

When sending a message, the routing key precedence is:

1. AmqpStamp::getRoutingKey() — message-specific routing key (highest priority)
2. default_publish_routing_key — configured default for publishing
3. '' — empty string (no routing key)

This separation prevents unintended coupling: setting routing_key for consumer binding does not affect how messages are published.

Pitfall — consumer binding vs sender stamp: because the consumer and sender keys are resolved independently, stamping a message with AmqpStamp::getRoutingKey() (or setting default_publish_routing_key) that differs from the queue's routing_key binding causes the broker to drop every unroutable message on a direct exchange — publish still reports success. When you publish with a per-message AmqpStamp, make sure your routing_key DSN option binds the queue with that same key (e.g. publish with new AmqpStamp('test') and configure ...?queue=test&routing_key=test).

Multi-queue mode

A single transport can consume several queues on one shared channel. Because the channel is shared, max_unacked_messages is a total budget (divided across the queues) and an idle get() over N queues costs about one read_timeout, not N (#239, #309).

Programmatic configuration:

$options = [
    'queues' => [
        'orders' => ['binding_keys' => ['order.created', 'order.updated']],
        'payments' => ['binding_keys' => ['payment.*']],
        'dead_letters' => ['arguments' => ['x-message-ttl' => 60000]],
    ],
    'batch_size' => 10,
];

The same via the DSN query string:

amqp-consoomer://guest:guest@localhost:5672/%2f/events?queues[orders][binding_keys][0]=order.created&queues[payments][binding_keys][0]=payment.*&batch_size=10

Each entry under queues supports binding_keys (list), binding_arguments (map) and arguments (map) for that queue. With queues, the single queue option is not used.

Fairness caveat (shared channel). All configured queues share one channel, and AMQP prefetch is per-consumer. Under a sustained backlog on one queue the broker keeps delivering that queue's messages, so the batch can fill before the other queues are polled — the round-robin start rotation spreads this across get() calls but does not eliminate it. For workloads where strict per-queue fairness matters, run one transport (hence one channel) per queue. Tracked as #374; a shared-channel fix would require per-queue channels or a pull model.

Delayed messages

Attach AmqpDelayStamp (delay in milliseconds) and the sender publishes to a per-delay queue whose message TTL expires it back onto the main exchange:

use CrazyGoat\TheConsoomer\AmqpDelayStamp;
use Symfony\Component\Messenger\Envelope;

$transport->send(new Envelope($message, [new AmqpDelayStamp(30_000)]));

The sender declares a durable delay exchange (delay.exchange_name) and, per distinct (delay, routing key) pair, a delay queue named by delay.queue_name_pattern (default delay_{delay}_{queue}) with x-message-ttl set to the delay. The delayed message is published with its own routing key and the queue is bound for that key, while the queue carries only x-dead-letter-exchange — so on expiry the message keeps its original routing key and returns to the main exchange correctly. This also holds for custom patterns without {queue}, where several routing keys share one delay queue (#276). The routing key used here is validated: only A-Z a-z 0-9 . _ - are allowed and it must fit the AMQP 255-byte name limit, otherwise send() throws InvalidArgumentException (#289).

The set of declared delay queues (and their bindings) is cached in-process and bounded to delay.max_tracked_queues entries (default 1000) so that a worker publishing many distinct (delay, routing key) pairs — e.g. per-message delays — does not grow memory for the process lifetime. The cache is only an optimisation: once an entry is evicted the next publish re-declares the queue and re-binds the key, both idempotent on the broker. Note that the broker-side queues themselves are durable and are not removed on eviction; draw delays from a small, fixed set of levels to avoid broker-side queue proliferation (#211).

Message priorities

Attach AmqpPriorityStamp to publish with a priority. The target queue must be declared with x-max-priority, e.g. queue_arguments[x-max-priority]=10:

use CrazyGoat\TheConsoomer\AmqpPriorityStamp;

$transport->send(new Envelope($message, [new AmqpPriorityStamp(9)]));

Stamps

Send-side stamps are NonSendableStampInterface, so a received envelope never carries them back into a retry/redelivery.

Reconnect and redelivery

AMQP delivery tags are scoped to a channel. When a channel is lost (broker restart, dropped socket, heartbeat-stale reconnect), the broker redelivers its in-flight messages on the next channel. The receiver tracks a channel generation on every received envelope, so ack()/reject() of an envelope from a lost channel is a no-op (the broker redelivers it) rather than a protocol error that would close the channel (#220). ack()/reject() never trigger a reconnect themselves (#235) — see Heartbeat.

Heartbeat

The heartbeat feature keeps connections alive and detects dead connections. When enabled, the connection will automatically reconnect if no activity is detected for twice the heartbeat interval.

framework:
    messenger:
        transports:
            consoomer:
                dsn: 'amqp-consoomer://guest:guest@localhost:5672/%2f/messages?queue=my_queue&heartbeat=60'

With heartbeat enabled:

• The staleness check (elapsed > 2 * heartbeat, measured from the last application activity) runs before send() and get(), and before queue-management calls. If stale, the channel is renewed.
• ack()/reject() deliberately do not reconnect on staleness (#235): they run after the (possibly slow) message handler, and reconnecting there would wipe in-flight delivery tags and redeliver messages. They refresh the activity timestamp instead, so a slow-but-healthy handler does not cause a reconnect on the next get().
• Activity is updated at the start and end of the consume cycle and around ack/reject.
• This wall-clock staleness is not a liveness probe — the broker-negotiated AMQP heartbeat on the native connection detects a genuinely dead socket.
• In-flight messages whose channel was genuinely lost are not acknowledged on the new channel — their delivery tag belongs to the dead channel, so ack/reject become no-ops and the broker redelivers them on the next get()

SSL/TLS

For TLS-encrypted connections use the amqps-consoomer:// scheme (or the legacy amqps://, which emits a deprecation notice). SSL options are passed as DSN query parameters or programmatic options:

Pitfall — ssl_verify without ssl_cacert: when verification is on (the default) but no CA certificate is pinned, the connection falls back to the system CA store. On builds where the system has no trusted CAs the handshake fails, or on some builds it verifies against an empty trust set — both silently. AmqpFactory::configureSsl() emits a prominent warning in this case — through the configured logger, or as a PHP E_USER_WARNING when no logger is wired (#351); set ssl_cacert explicitly in production to pin the broker's CA.

Security (#361): ssl_verify=false disables TLS peer-certificate verification, enabling man-in-the-middle / broker impersonation. It is refused unless you set "allow_insecure_verify" => true in the programmatic transport options — it cannot be set from the DSN query string, so a config-file or env-var DSN can never self-authorize the downgrade. Use it only for local development with self-signed certificates.

framework:
    messenger:
        transports:
            consoomer:
                dsn: 'amqps-consoomer://guest:guest@rabbit:5671/%2f/messages?queue=my_queue&ssl_cacert=/etc/ssl/certs/rabbit-ca.pem'

Retry Configuration

The transport supports configurable retry logic with exponential backoff, jitter, and circuit breaker.

framework:
    messenger:
        transports:
            consoomer:
                dsn: 'amqp-consoomer://guest:guest@localhost:5672/%2f/messages?queue=my_queue&retry=1&retry_count=3&retry_delay=500000&retry_backoff=1&retry_jitter=1&retry_circuit_breaker=1'

With retry enabled:

• Connection and channel failures are retried automatically up to retry_count times (including the first attempt)
• Non-AMQP exceptions are not retried
• Permanent failures are identified by reply code, not by exception type: 403 (access refused), 404 (not found), 405 (resource locked) and 406 (precondition failed), read from AMQPException::getCode() or the symbolic name in the broker message (ACCESS_REFUSED/NOT_FOUND/RESOURCE_LOCKED/PRECONDITION_FAILED), are not retried. Everything else is retried, including AMQPQueueException/AMQPExchangeException without a permanent reply code — ext-amqp raises AMQPQueueException for a plain read timeout, which is transient. Connection/channel errors (AMQPConnectionException, AMQPChannelException) are always retried.
• On exhaustion, a RetryExhaustedException is thrown with the last failure as previous

Circuit Breaker Scope

The circuit-breaker state is per-process and in-memory only. There is no shared store behind it, so its protection is limited to the lifetime of the current process:

• PHP-FPM / request-scoped senders: a new CircuitBreaker instance is created on every request, so failure counters reset each time and a high retry_circuit_breaker_threshold is effectively never reached. Only failures that recur within a single request count.
• Worker fleet (messenger:consume): each consumer process has its own independent breaker. A broker-wide outage is not coordinated across the fleet — every process must trip (and recover) on its own.
• Long-lived single workers: this is where the breaker works as documented — counters persist between operations and the circuit opens after retry_circuit_breaker_threshold consecutive failures.

To size the breaker usefully, keep the effective failure budget in mind: with the defaults (retry_count=3 attempts per operation) an OPEN circuit is reached after roughly retry_circuit_breaker_threshold / retry_count consecutive failed operations — by default 10 / 3, i.e. after about 3–4 consecutive failing operations.

No shared-state backend (APCu, Redis, …) is currently provided; do not treat retry_circuit_breaker_* as fleet-wide or request-spanning protection.

Untrusted brokers and poison messages

The raw AMQP body is broker-controlled input: anyone who can publish to (or impersonate) a consumed queue decides what bytes reach your serializer. Two guards apply on the receive path (#288):

• Size guard — a body larger than max_body_bytes is rejected without ever being handed to the serializer, so a single oversized publish cannot push the consumer into memory pressure. Set max_body_bytes: 0 to disable the guard.
• Poison-pill containment — when the serializer rejects a body (raises MessageDecodingFailedException), the offending message is rejected (dropped or dead-lettered per broker policy) and the current batch keeps flowing. If the message cannot even be rejected (broken channel), the decode failure propagates so the problem stays visible instead of looping silently on broker redelivery.

Warning: do not use PhpSerializer with untrusted publishers. Symfony's PhpSerializer runs unserialize() on the raw body. That is PHP object injection and gadget-chain RCE, not a data error — no size limit changes that. It is a supported option only for first-party publishers you fully trust; with anything untrusted, use a data-format serializer (JSON) and validate on the application side.

Publish Reliability

Warning: without publisher confirms, publishes are fire-and-forget. AMQPExchange::publish() writes to the socket buffer and returns immediately. If the broker is down, the exchange is missing, or the topology was lost after a restart, send() still reports success and the message is silently lost. The retry mechanism is inert on the send path without an error signal.

Publisher confirms are controlled by publisher_confirms; confirm_timeout is only the wait duration. publisher_confirms defaults to true when a positive confirm_timeout is configured (backwards compatible) and false otherwise, so set it explicitly to be unambiguous. confirm_timeout=0 now means wait indefinitely, matching ext-amqp's waitForConfirm(0).

Enable publisher confirms for reliable publishing (e.g. publisher_confirms=1&confirm_timeout=5):

framework:
    messenger:
        transports:
            consoomer:
                dsn: 'amqp-consoomer://guest:guest@localhost:5672/%2f/messages?queue=my_queue&confirm_timeout=5&retry=1'

With confirm_timeout enabled:

• The sender calls confirmSelect() on the channel (once per channel, cached) and waitForConfirm() after each publish
• The broker acknowledges each publish; a 404 (missing exchange) or nack surfaces as an AMQPException
• Combined with retry=1, transient failures are retried automatically

With retry=1 enabled (regardless of confirms):

• The sender checks isConnected() before each publish attempt and reconnects if the broker is down, giving the retry wrapper an error signal to act on

With auto_setup=true (default):

• A producer declares only its exchange on send; queues and bindings are consumer-side topology and are declared by the receiver
• With retry enabled, the declaration runs inside the retried operation, so a transient declaration failure is retried instead of escaping send() (the connection heartbeat is checked before setup, too)
• Topology is not re-declared on every reconnect by default — durable exchanges/queues survive disconnects. Set redeclare_on_reconnect=true to re-declare after a reconnect (useful when an operator may delete the topology while the worker is connected); Sender::ensureConnected()/Receiver::ensureConnected() then reset the setup flag — including after a genuine channel failure detected during get()
• If you declare non-durable topology (durable=false), a broker restart deletes it and it will not be recreated unless redeclare_on_reconnect=true: durable is the default precisely because it survives disconnects

Testing

Run tests

# All tests
composer test

# Unit tests only
composer test-unit

# E2E tests (requires RabbitMQ)
composer test-e2e-full

E2E tests

E2E tests require RabbitMQ. The test-e2e-full script automatically:

1. Starts RabbitMQ via Docker
2. Waits for RabbitMQ to be ready
3. Runs E2E tests
4. Stops RabbitMQ

Code quality

# Check everything: phpstan + rector + php-cs-fixer (dry-run; fixes nothing)
composer lint

# Apply fixes: rector then php-cs-fixer
composer lint:fix

Contributing

Contributions are welcome! Please open an issue or submit a pull request.

License

This project is licensed under the MIT License.

Links

• GitHub Repository
• Symfony Messenger Documentation