Rabbitmq Bundle Laravel Package

## Getting Started

### Minimal Setup
1. **Installation**:
   ```bash
   composer require php-amqplib/rabbitmq-bundle

Enable the bundle in config/bundles.php:

return [
    // ...
    PhpAmqpLib\RabbitMqBundle\RabbitMqBundle::class => ['all' => true],
];

2. Configuration: Add RabbitMQ connection details to config/packages/rabbitmq.yaml:

   rabbit_mq:
       hosts:
           default:
               host:     'localhost'
               port:     5672
               user:     'guest'
               password: 'guest'
               vhost:    '/'
       connections:
           default: ~
       producers:
           default: ~
       consumers:
           default: ~
   

3. First Use Case: Publish a message from a controller:

   use PhpAmqpLib\RabbitMqBundle\RabbitMq\Producer;
   
   class MessageController extends AbstractController
   {
       public function send(Producer $producer): Response
       {
           $producer->publish(
               'message',
               'Hello RabbitMQ!'
           );
           return new Response('Message sent!');
       }
   }
   

4. Consuming Messages: Create a consumer service (e.g., src/Message/MessageConsumer):

   use PhpAmqpLib\RabbitMqBundle\RabbitMq\ConsumerInterface;
   
   class MessageConsumer implements ConsumerInterface
   {
       public function execute(array $body, string $json): void
       {
           // Process message
       }
   }
   

   Register it in config/packages/rabbitmq.yaml:

   consumers:
       default:
           callback: 'service:message_consumer'
           queue: 'message'
           exchange: 'message'
           routing_key: 'message'
   

Implementation Patterns

Common Workflows

1. Publishing Messages

• Direct Publishing:

  $producer->publish('exchange', 'routing_key', $message, $properties = []);
  

  Use for simple fire-and-forget messaging.

• Delayed Messages (via x-delay header):

  # config/packages/rabbitmq.yaml
  producers:
      delayed:
          exchange: 'delayed'
          routing_key: 'delayed.key'
          headers:
              'x-delay': 10000 # 10 seconds
  

  $producer->publish('delayed', $message);
  

• Batch Publishing:

  $producer->publishBatch('exchange', 'routing_key', [$msg1, $msg2, $msg3]);
  

2. Consuming Messages

• Asynchronous Consumption: Use Symfony’s Messenger component to decouple consumers:

  # config/packages/messenger.yaml
  framework:
      messenger:
          transports:
              rabbitmq: '%env(RABBITMQ_DSN)%'
          routing:
              'App\Message\ProcessMessage': rabbitmq
  

  // In a consumer service
  use Symfony\Component\Messenger\MessageBusInterface;
  
  class AsyncConsumer
  {
      public function __construct(private MessageBusInterface $bus)
      {}
  
      public function execute(array $body, string $json): void
      {
          $this->bus->dispatch(new ProcessMessage($json));
      }
  }
  

• Prefetch Count (limit unacknowledged messages):

  consumers:
      default:
          prefetch_count: 10
  

3. RPC (Remote Procedure Call)

• Request-Reply Pattern:

  $rpc = new \PhpAmqpLib\RabbitMqBundle\RabbitMq\Rpc\RpcClient($connection);
  $result = $rpc->call('rpc_queue', 'Hello', 1000); // Timeout: 1s
  

4. Fanout Exchanges

• Broadcast Messages:

  producers:
      fanout:
          exchange: 'fanout_exchange'
          exchange_type: 'fanout'
  

  $producer->publish('fanout', '', $message); // No routing key
  

5. Topic Exchanges

• Dynamic Routing:

  producers:
      topic:
          exchange: 'topic_exchange'
          exchange_type: 'topic'
  

  $producer->publish('topic', 'logs.error', $message);
  

Integration Tips

Symfony Services

• Dependency Injection: Inject Producer or ConsumerInterface directly into services:

  public function __construct(private Producer $producer) {}
  

• Event Listeners: Trigger messages on events:

  use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  
  class OrderEventSubscriber implements EventSubscriberInterface
  {
      public function __construct(private Producer $producer) {}
  
      public static function getSubscribedEvents(): array
      {
          return [OrderCreatedEvent::class => 'onOrderCreated'];
      }
  
      public function onOrderCreated(OrderCreatedEvent $event): void
      {
          $this->producer->publish('order_events', 'order.created', $event->getOrder());
      }
  }
  

Retry Logic

• Dead Letter Exchanges (DLX): Configure in rabbitmq.yaml:

  consumers:
      default:
          queue:
              dead_letter_exchange: 'dead_letter'
              dead_letter_routing_key: 'dead.letter'
  

  Use for failed message recovery.

Monitoring

• Metrics Integration: Use symfony/monolog-bundle to log message flow:

  $this->logger->info('Message published', ['exchange' => 'exchange', 'routing_key' => 'key']);
  

Gotchas and Tips

Pitfalls

1. Connection Management

• Connection Drops: RabbitMQ connections can drop silently. Use reconnect config:

  rabbit_mq:
      hosts:
          default:
              reconnect: true
              reconnect_delay: 5000 # 5s
  

  • Tip: Implement a health check endpoint to verify connectivity.

2. Consumer Lifecycle

• Consumer Stuck in "Running" State: If consumers hang, check for:
  • Unhandled exceptions (wrap execute() in try-catch).
  • Infinite loops in message processing.
  • Fix: Use ack()/nack() explicitly:

    public function execute(array $body, string $json): void
    {
        try {
            // Process
            $this->getConsumer()->ack();
        } catch (\Exception $e) {
            $this->getConsumer()->nack(false, false, 'Error: ' . $e->getMessage());
        }
    }
    

3. Serialization Issues

• Non-Serializable Objects: RabbitMQ requires messages to be serializable. Use json_encode() or a serializer:

  $producer->publish('exchange', 'key', json_encode($object));
  

  • Tip: Extend the bundle’s MessageSerializer for custom logic.

4. Queue Binding Conflicts

• Duplicate Bindings: If a queue is bound multiple times to the same exchange/key, messages may be processed twice.
  • Fix: Use unique queue names or idempotent consumers.

5. Memory Leaks

• Unclosed Channels: Always ensure channels/connections are closed in onShutdown():

  public function onShutdown(): void
  {
      $this->getConnection()->close();
  }
  

Debugging Tips

1. Enable Logging

• Verbose Logs:

  rabbit_mq:
      logging: true
  

  Logs will appear in var/log/dev.log.

2. RabbitMQ Management UI

• Inspect Queues/Exchanges: Access http://localhost:15672 (default credentials: guest/guest).
  • Check for stuck messages, unacked messages, or binding errors.

3. Common Errors

4. Testing

• Unit Testing Consumers: Mock the ConsumerInterface:

  $consumer = $this->createMock(Consumer