Getting Started

Minimal Setup

Install the Package
```
composer require symfony/amqp-messenger
```
Ensure ext-amqp is enabled in your PHP environment (required for AMQP support).

Configure RabbitMQ Transport Add to config/messenger.php:

'transports' => [
    'amqp' => [
        'dsn' => 'amqp://user:pass@localhost:5672/%2f/messages',
        'options' => [
            'queue' => 'laravel_tasks',
            'exchange_options' => ['name' => 'laravel_tasks_exchange'],
            'queues' => [
                'laravel_tasks' => [
                    'queue_name' => 'laravel_tasks',
                    'exchange' => [
                        'name' => 'laravel_tasks_exchange',
                        'type' => 'direct',
                    ],
                ],
            ],
        ],
    ],
],

Dispatch a Message

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;

$bus->dispatch(new YourMessage(), [
    new DelayStamp(3600), // Delay for 1 hour
]);

Run the Worker

php bin/console messenger:consume amqp -vv

First Use Case: Delayed Job Processing

Replace Laravel’s delay() with Symfony’s DelayStamp for time-based job execution:

// In a controller or command
$bus->dispatch(new ProcessPayment($orderId), [
    new DelayStamp(60), // 60 seconds delay
]);

Implementation Patterns

Core Workflows

1. Delayed Message Dispatch

Use DelayStamp for time-based processing (supports seconds, minutes, hours, or DateTimeInterface):

// Delay for 5 minutes
$bus->dispatch(new SendEmail(), [
    new DelayStamp(300),
]);

// Delay until a specific time
$delayUntil = new \DateTime('+1 day');
$bus->dispatch(new GenerateReport(), [
    new DelayStamp($delayUntil),
]);

2. Exchange-to-Exchange Routing

Leverage AMQP exchanges for advanced routing (e.g., fanout, topic):

// Configure in config/messenger.php
'exchange_options' => [
    'name' => 'notifications',
    'type' => 'topic',
],

// Bind queues to exchanges
'queues' => [
    'emails' => [
        'exchange' => ['name' => 'notifications', 'routing_key' => 'email.*'],
    ],
    'sms' => [
        'exchange' => ['name' => 'notifications', 'routing_key' => 'sms.*'],
    ],
],

3. Failure Handling

Route failed messages to a dead-letter queue (DLQ):

// In config/messenger.php
'failure_transport' => 'failed',
'transports' => [
    'failed' => [
        'dsn' => 'amqp://user:pass@localhost:5672/%2f/failed',
        'options' => [
            'queue' => 'failed_tasks',
            'exchange_options' => ['name' => 'failed_tasks_exchange'],
        ],
    ],
],

4. Worker Scaling

Run multiple workers for parallel processing:

# Terminal 1
php bin/console messenger:consume amqp -vv --limit=10

# Terminal 2 (for delayed messages)
php bin/console messenger:consume amqp --time-limit=3600 -vv

Integration Tips

Laravel-Specific Adjustments

Service Provider Setup Register the transport in AppServiceProvider:

public function boot()
{
    $this->app->make(\Symfony\Component\Messenger\MessageBus::class)
        ->addTransport('amqp', new \Symfony\Component\Messenger\Transport\Serialization\Serializer());
}

Command Bus Alias Override Laravel’s Bus facade to use Symfony’s MessageBus:

// config/app.php
'aliases' => [
    'Bus' => \Symfony\Component\Messenger\MessageBus::class,
],

Artisan Command Wrapper Create a custom command to dispatch messages:

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Symfony\Component\Messenger\MessageBusInterface;

class DispatchMessage extends Command
{
    protected $signature = 'message:dispatch {message} {--delay=0}';
    protected $description = 'Dispatch a message with optional delay';

    public function handle(MessageBusInterface $bus)
    {
        $bus->dispatch(new \App\Messages\YourMessage(), [
            new \Symfony\Component\Messenger\Stamp\DelayStamp((int) $this->option('delay')),
        ]);
    }
}

Performance Optimization

Batch Processing: Use --limit in worker commands to process messages in batches.
Prefetch Count: Adjust RabbitMQ prefetch count to control message load per worker:
```
'options' => [
    'prefetch_count' => 50, // Default is 1
],
```
Connection Pooling: Reuse connections for high-throughput scenarios by configuring a connection factory.

Gotchas and Tips

Pitfalls

RabbitMQ Version Mismatch
- Delayed Quorum Queues: Require RabbitMQ ≥4.0. Older versions will throw errors.
- Quorum Queues: Require RabbitMQ ≥3.11. Test with:
```
rabbitmqctl status | grep -i quorum
```
- Fix: Downgrade to classic delayed exchanges if using RabbitMQ <4.0:
```
'options' => [
    'use_delayed_exchange' => true, // Fallback for RabbitMQ <4.0
],
```
Negative Delay Values
- Issue: Passing negative delays (e.g., -60 for "send immediately") may cause unexpected behavior.
- Fix: Validate delays in your message handlers or use 0 for immediate dispatch.
Heartbeat Timeouts
- Issue: Long-running message handlers may trigger connection timeouts (default: 60s).
- Fix: Adjust heartbeat settings in the DSN or options:
```
'options' => [
    'heartbeat' => 30, // 30-second heartbeat
],
```

Quorum Queue Disk Usage

Issue: Quorum queues replicate data across nodes, increasing disk usage.

Fix: Monitor disk space and adjust quorum_queue_node_count:

'options' => [
    'queues' => [
        'your_queue' => [
            'quorum_queue_node_count' => 2, // Reduce replication factor
        ],
    ],
],

Message Serialization
- Issue: AMQP has a 16MB message size limit. Large payloads (e.g., files) will fail.
- Fix: Offload large data to storage (e.g., S3) and pass references:
```
$bus->dispatch(new ProcessLargeFile($fileId));
```
Worker Stuck on Messages
- Issue: Workers may hang if messages aren’t acknowledged or fail silently.
- Fix: Enable debug logging and use --time-limit to force timeouts:
```
php bin/console messenger:consume amqp -vv --time-limit=300
```

Debugging Tips

Check RabbitMQ Queues Use rabbitmqctl to inspect queues:

rabbitmqctl list_queues name messages consumers
rabbitmqctl get_queue 'laravel_tasks'

Enable AMQP Debugging Add to config/messenger.php:

'transports' => [
    'amqp' => [
        'dsn' => 'amqp://user:pass@localhost:5672/%2f/messages?logger=debug',
    ],
],

Inspect Failed Messages Consume the failure transport manually:
```
php bin/console messenger:consume failed -vv
```

Test Delayed Messages Verify delays work as expected:

// Dispatch a test message
$bus->dispatch(new TestMessage(), [
    new DelayStamp(10), // 10 seconds
]);

// Check RabbitMQ for delayed messages
rabbitmqctl list_exchanges name type

Extension Points

Custom Transport Options Extend the transport configuration for advanced use cases:

'options' => [
    'connection' => [
        'host' => 'rabbitmq',
        'port' => 5672,
        'user' => 'guest',
        'password' => '

Amqp Messenger Laravel Package