Getting Started

Minimal Setup

Installation:

composer require php-amqplib/rabbitmq-bundle

new OldSound\RabbitMqBundle\OldSoundRabbitMqBundle(),

Configure RabbitMQ (config/packages/old_sound_rabbit_mq.yaml):

old_sound_rabbit_mq:
    connections:
        default:
            host:     'localhost'
            port:     5672
            user:     'guest'
            password: 'guest'
            vhost:    '/'
            lazy:     false
    producers:
        upload_picture_producer:
            connection:       default
            exchange_options: {name: 'upload_pictures', type: direct}
    consumers:
        upload_picture_consumer:
            connection:       default
            exchange_options: {name: 'upload_pictures', type: direct}
            queue_options:    {name: 'upload_pictures'}
            callback:         upload_picture_consumer

First Producer Use Case: Create a service to publish messages (e.g., src/Service/UploadPictureProducer.php):

namespace App\Service;

use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;

class UploadPictureProducer
{
    private $producer;

    public function __construct(ProducerInterface $producer)
    {
        $this->producer = $producer;
    }

    public function publish(array $message)
    {
        $this->producer->publish(serialize($message));
    }
}

services:
    App\Service\UploadPictureProducer:
        arguments:
            - '@old_sound_rabbit_mq.upload_picture_producer'

First Consumer Use Case: Create a consumer class (e.g., src/MessageHandler/UploadPictureConsumer.php):

namespace App\MessageHandler;

use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;

class UploadPictureConsumer implements ConsumerInterface
{
    public function execute(array $message)
    {
        $data = unserialize($message['body']);
        // Process $data['user_id'] and $data['image_path']
    }
}

Implementation Patterns

Workflows

Producer Workflow:

Symfony Controller:

use App\Service\UploadPictureProducer;

class UploadController extends AbstractController
{
    public function uploadPicture(UploadPictureProducer $producer)
    {
        $producer->publish(['user_id' => 123, 'image_path' => '/path/to/image.jpg']);
        return new Response('Message published!');
    }
}

Command Bus: Use Symfony Messenger or a custom command to decouple producers from controllers.

Consumer Workflow:

CLI Consumption:

php bin/console rabbitmq:consumer -m 50 upload_picture_consumer

Supervisor Integration: Configure a Supervisor process to run consumers as daemons:

[program:upload_picture_consumer]
command=php /path/to/your/project/bin/console rabbitmq:consumer upload_picture_consumer
autostart=true
autorestart=true
user=www-data
numprocs=1

Message Serialization:

Use serialize()/unserialize() for simplicity (as shown in README).

For complex objects, leverage Symfony Serializer:

# config/packages/old_sound_rabbit_mq.yaml
old_sound_rabbit_mq:
    producers:
        upload_picture_producer:
            serializer: serializer

use Symfony\Component\Serializer\SerializerInterface;

class UploadPictureProducer
{
    public function __construct(
        ProducerInterface $producer,
        private SerializerInterface $serializer
    ) {}

    public function publish(array $message)
    {
        $this->producer->publish(
            $this->serializer->serialize($message, 'json')
        );
    }
}

Error Handling:

Dead Letter Exchanges (DLX): Configure in consumer section:

consumers:
    upload_picture_consumer:
        ...
        queue_options:
            name: 'upload_pictures'
            dead_letter_exchange: 'dead_letter_exchange'

Retry Logic: Implement a custom ConsumerInterface with retry logic:

class RetryConsumer implements ConsumerInterface
{
    private $maxRetries = 3;

    public function execute(array $message)
    {
        try {
            $this->realConsumer->execute($message);
        } catch (\Exception $e) {
            if ($this->maxRetries-- > 0) {
                // Re-publish to DLX or retry queue
            }
            throw $e;
        }
    }
}

Dynamic Routing:

Use routing keys for dynamic message handling:

producers:
    dynamic_producer:
        exchange_options: {name: 'dynamic_exchange', type: direct}

$producer->publish(serialize($message), 'routing.key');

Consumers bind queues to routing keys:

consumers:
    dynamic_consumer:
        queue_options: {name: 'dynamic_queue'}
        exchange_options: {name: 'dynamic_exchange', type: direct}
        binding_options: {routing_key: 'routing.key'}

Gotchas and Tips

Pitfalls

Connection Management:
- Lazy Connections: Set lazy: true in config to avoid connection overhead for infrequent producers.
- Connection Failures: Implement a circuit breaker pattern for consumers (e.g., using php-amqplib's Connection events).
Message Ordering:
- RabbitMQ does not guarantee message order by default. Use priority queues or separate queues per priority if ordering is critical:
```
queue_options:
    name: 'priority_queue'
    flags: {x-max-priority: 10}
```
Memory Leaks:
- Unserialization: Avoid unserialize() for untrusted data (security risk). Use json_decode() or a custom serializer.
- Consumer Processes: Ensure consumers release resources (e.g., close channels) after processing:
```
public function execute(array $message)
{
    try {
        // Process message
    } finally {
        $this->getChannel()->close();
    }
}
```

Configuration Overrides:

Environment-Specific Config: Use Symfony's parameter system to override RabbitMQ settings per environment:

# config/packages/old_sound_rabbit_mq.yaml
old_sound_rabbit_mq:
    connections:
        default:
            host: '%env(RABBITMQ_HOST)%'

export RABBITMQ_HOST=prod-rabbitmq.example.com

Consumer Lifecycle:

Graceful Shutdown: Consumers may hang on get() if not handled properly. Use set_blocking(false) for non-blocking calls or implement a timeout:

$channel->basic_consume($consumerTag, 'queue', false, false, false, false, [$this, 'execute']);
if (!$channel->is_consuming()) {
    $channel->basic_consume($consumerTag, 'queue', false, false, false, false, [$this, 'execute'], ['timeout' => 5]);
}

Debugging Tips

RabbitMQ Management Plugin:
- Enable the plugin (rabbitmq-plugins enable rabbitmq_management) and access the UI at http://localhost:15672 (default credentials: guest/guest).
- Monitor queues, messages, and consumer activity in real-time.

Logging:

Enable debug logging for old_sound_rabbit_mq in config/packages/monolog.yaml:

handlers:
    rabbitmq:
        type: stream
        path: "%kernel.logs_dir%/%kernel.environment%.rabbitmq.log"
        level: debug
        channels: ["old_sound_rabbit_mq"]

Consumer Debugging:
- Manual Consumption: Test consumers manually with -v (verbose) flag:
```
php bin/console rabbitmq:consumer -v upload_picture_consumer
```
- Simulate Failures: Inject exceptions in execute() to test error handling.
Common Errors:
- ConnectionException: Verify RabbitMQ server is running and credentials are correct.
- NotFoundException: Check if the exchange/queue exists and the consumer is bound correctly.
- AccessRefused: Ensure the user has permissions for the vhost (e.g., `rabbitmqctl set_permissions -p /

Rabbitmq Bundle Laravel Package