Getting Started

Install the tools alongside a transport and (for Laravel) the queue bridge:

composer require enqueue/amqp-tools enqueue/amqp-lib enqueue/laravel-queue
# or use enqueue/amqp-ext if ext-amqp is compiled

enqueue/amqp-tools is not a transport — it decorates an AmqpContext from amqp-lib, amqp-ext, or amqp-bunny. The moment you have a working enqueue AMQP connection, this package is what unlocks the "extras" the AMQP 0-9-1 spec doesn't give you: delayed delivery, per-message priority, TTL, max-length policies.

Where to look first (in this order):

Interop\Amqp\AmqpContext — the contract you get back from the transport.
Enqueue\AmqpTools\DelayStrategy and its two implementations:
- RabbitMqDlxDelayStrategy (DLX + TTL, no broker plugin needed — the default).
- RabbitMqDelayPluginDelayStrategy (requires the x-delayed-message plugin).
Enqueue\AmqpTools\RabbitMqPrioDeliveryStrategy for priority queues.
Interop\Amqp\AmqpMessage::PROPERTY_HEADER — headers are where AMQP extensions live (x-delay, x-priority).

First use case: a delayed job. Dispatch a follow-up email that runs in 30 minutes without spinning up a scheduler entry per job:

// app/Jobs/SendFollowUpEmail.php implements ShouldQueue
SendFollowUpEmail::dispatch($user)->onConnection('amqp')->delay(now()->addMinutes(30));

That single call exercises everything the package is for — the tools translate delay() into the correct broker-native mechanism behind the scenes.

Baseline config in config/queue.php:

'amqp' => [
    'driver' => 'amqp',
    'host'   => env('AMQP_HOST', 'localhost'),
    'port'   => env('AMQP_PORT', 5672),
    'login'  => env('AMQP_USER', 'guest'),
    'password' => env('AMQP_PASSWORD', 'guest'),
    'vhost'  => env('AMQP_VHOST', '/'),
    'exchange' => ['name' => 'laravel', 'type' => 'direct'],
    'queue' => env('AMQP_QUEUE', 'default'),
    'delay_strategy' => 'dlx',          // 'dlx' | 'plugin'
    'prio_delivery_strategy' => 'rabbitmq',
    'max_priority' => 10,
],

Then QUEUE_CONNECTION=amqp and you're running jobs over RabbitMQ with php artisan queue:work amqp --queue=default.

Implementation Patterns

1. Delayed dispatch inside jobs

class ProcessOrder implements ShouldQueue
{
    public function handle(): void
    {
        // ...
        // Retry this job in 5 minutes without manual retry counting
        $this->release(300);
    }
}

release() on an AMQP connection flows through DelayStrategy and re-publishes with the correct x-delay header or DLX TTL. The same plumbing backs ->delay() at dispatch time; learn one, you get both.

2. Priority lanes via `onQueue()`

// config/queue.php — declare priority on the connection
'prio_delivery_strategy' => 'rabbitmq',
'max_priority' => 10,

// dispatch
ImportantReport::dispatch($report)->onQueue('reports')->onConnection('amqp');

The package sets x-priority on the message and (crucially) re-declares the queue with x-max-priority so the broker honours it. Workers need a matching --queue=reports and a prefetch of 1–5 for fair priority behaviour.

3. Combined delay + priority

CriticalAlert::dispatch($alert)
    ->onConnection('amqp')
    ->onQueue('alerts')
    ->delay(now()->addSeconds(30));

Both strategies cooperate: the message gets x-priority and the delay header. If you need this at scale, add 'prio_delivery_strategy' => 'rabbitmq' plus 'delay_strategy' => 'dlx' — mixing plugin-delay with priority is fragile because the plugin re-injects messages at the exchange level.

4. Non-Laravel producer, Laravel consumer

A common microservice shape: a Node/Python service publishes into a topic, Laravel jobs consume it.

$context = AmqpContext::create('amqp://guest:guest@rabbit:5672/%2f');
$context->setDelayStrategy(new RabbitMqDlxDelayStrategy());

$queue = $context->createQueue('jobs.email');
$context->declareQueue($queue);

$consumer = $context->createConsumer($queue);
$message = $consumer->receive(5000);
if ($message) {
    try {
        dispatch(unserialize($message->getBody()))->onConnection('sync');
        $consumer->acknowledge($message);
    } catch (\Throwable $e) {
        $consumer->reject($message, true); // requeue
    }
}

Wrap this in a long-running Artisan command and you have a bespoke worker for legacy producers — the tools still handle delay/priority on any message you republish.

5. Custom `DelayStrategy`

When you want a different backoff policy than the plugin/DLX defaults (e.g., a single shared retry ladder queue):

class LadderDelayStrategy implements DelayStrategy
{
    public function delayMessage(AmqpMessage $message, AmqpDestination $dest, $delay): void
    {
        // route through your own retry.{10s,1m,10m} topics
        $topic = /* ... */ $message->setHeader('x-delay', (int) $delay);
        // publish via a producer obtained from the context
    }
}

Then $context->setDelayStrategy(new LadderDelayStrategy()) in a service provider. This is the extension point most teams end up touching once their retry semantics diverge from Laravel's exponential default.

6. Testing

AmqpContext is an interface — mock it (Mockery/Pest) rather than hitting a live broker in unit tests. For integration tests, use enqueue/amqp-lib against a Dockerised RabbitMQ with testcontainers or the queue:fake-style helper if you're on enqueue/laravel-queue's newer versions. Assert on AmqpMessage::getHeaders() to verify delay/priority wiring without touching a real broker.

Gotchas and Tips

The DLX strategy is a queue factory. Each distinct delay value creates a transient queue (delay.{ttl}.{target}). A job that delays 3 different durations creates 3 queues; a job that delays random_int() creates hundreds. Cap your delay granularity (10s, 1m, 5m, 1h) or switch to the plugin. Watch the RabbitMQ management UI for queue churn.

The delay plugin needs to be installed and configured. x-delayed-message is not bundled with RabbitMQ. If delay_strategy: plugin is set but the plugin is missing, exchanges silently fail to declare — you'll see messages vanish and wonder why ->delay() is a no-op. Verify with rabbitmq-plugins list | grep delayed before trusting it in prod.

x-delay is a header, not a property. AmqpMessage::PROPERTY_HEADER is where it lives; setting it via setProperty() won't work and won't error. Same for x-priority. Always inspect with getHeaders() when debugging.

Priority requires the queue to be declared with x-max-priority — before any message arrives. If the queue already exists with the default (no priority), sending x-priority is a silent no-op. Either delete and redeclare the queue during rollout, or use a versioned queue name (jobs.v2). This is the single biggest "it works in dev, not in prod" trap with this package.

Prefetch and priority fight each other. With prefetch_count=100, a worker drains all high-priority messages before the broker can interleave anything. Use 1–5 for priority queues, higher (20–50) for throughput-only queues. Configure via the options array in config/queue.php.

Don't double-retry. Laravel already implements retry/backoff on the job layer; if you also set DLX redelivery on the same queue, a failed job bounces between Laravel's counter and the broker's dead-letter policy. Pick one: either trust Laravel's $tries/$backoff, or drop ShouldQueue's retry and manage it purely in AMQP. The tools make it tempting to mix — resist.

Connection lifecycle matters. AmqpContext::create() per job is fine for CLI workers but catastrophic for HTTP dispatch (one TCP handshake per request). Register the context as a singleton in AppServiceProvider and let the enqueue/laravel-queue connector manage reconnection on AMQPConnectionClosedException.

amqp-lib vs amqp-ext vs amqp-bunny behave differently. x-delayed-message plugin support is best in amqp-lib and amqp-bunny; amqp-ext has historically lagged on some headers. If you rely on RabbitMqDelayPluginDelayStrategy, test against your chosen transport first — the interface is documented, the transport quirks are not.

Durability is opt-in. By default, queues are transient and messages non-persistent. In a Laravel job context you almost always want:

$queue->addFlag(AmqpQueue::FLAG_DURABLE);
$message->setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT);

Skip this and a broker restart silently drops your delayed and priority jobs — the two things you adopted the package for.

Publisher confirms are not on by default. For reliable dispatch from HTTP request handlers, enable confirms (transport-level option in amqp-lib). Otherwise a failed dispatch() looks successful.

Observability is your job. RabbitMQ's management UI is the primary dashboard; nothing in amqp-tools surfaces metrics. Add structured logging around acknowledge() / reject() in custom consumers, and export queue depth via the HTTP management API to your metrics stack (Prometheus/Datadog) — the package won't do it for you.

Beware max_priority drift between producer and consumer. If the producer's config says max_priority=10 and a consumer (or a teammate's .env) says 5, you get PRECONDITION_FAILED on channel open. Pin these values in a shared config constant, not two env files.

Amqp Tools Laravel Package