Rabbitmq Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require php-amqplib/rabbitmq-bundle
   

   Register the bundle in AppKernel.php:

   new OldSound\RabbitMqBundle\OldSoundRabbitMqBundle(),
   

2. 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: 'uploads', type: direct}
       consumers:
           upload_picture_consumer:
               connection:       default
               exchange_options: {name: 'uploads', type: direct}
               queue_options:    {name: 'upload.picture'}
               callback:         upload_picture_callback
   

3. First Producer Use Case: In a controller or service, inject and use the producer:

   use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
   
   class UploadController extends AbstractController
   {
       public function uploadPicture(ProducerInterface $producer): void
       {
           $msg = ['user_id' => 123, 'path' => '/image.jpg'];
           $producer->publish(serialize($msg), 'upload.picture');
       }
   }
   

4. First Consumer CLI Command: Run the consumer (50 messages at a time):

   php bin/console rabbitmq:consumer upload_picture_consumer -m 50
   

Implementation Patterns

Core Workflows

1. Producer-Consumer Pattern:

   • Producers: Publish messages to exchanges (e.g., upload_picture_producer).

     $producer->publish(serialize($data), 'routing_key');
     

   • Consumers: Process messages from queues (e.g., upload_picture_consumer). Define a callback in config/packages/old_sound_rabbit_mq.yaml:

     callback: App\Service\UploadPictureConsumer
     

     Implement the callback as a service:

     class UploadPictureConsumer implements ConsumerInterface
     {
         public function execute(Message $message)
         {
             $data = unserialize($message->body);
             // Process $data...
         }
     }
     

2. Message Serialization:

   • Use serialize()/unserialize() for simplicity.
   • For complex data, consider JSON or custom serializers (e.g., json_encode() + json_decode()).

3. Error Handling:

   • Wrap producer/publisher calls in try-catch:

     try {
         $producer->publish($msg);
     } catch (Exception $e) {
         $this->logger->error('RabbitMQ publish failed', ['error' => $e->getMessage()]);
     }
     

   • Consumers: Use nack() for failed messages (requeue or dead-letter):

     if (!$this->processMessage($message)) {
         $message->nack(false); // Reject and requeue
     }
     

4. Dynamic Routing:

   • Bind queues to exchanges dynamically (e.g., for multi-tenant apps):

     $connection = $this->get('old_sound_rabbit_mq.default_producer_connection');
     $channel = $connection->channel();
     $channel->queue_bind('queue_name', 'exchange_name', 'tenant_routing_key');
     

5. Background Processing:

   • Run consumers in the background (e.g., via Supervisor):

     php bin/console rabbitmq:consumer consumer_name --limit=0 --timeout=30
     

Integration Tips

1. Symfony Events:

   • Trigger RabbitMQ producers from Symfony events (e.g., kernel.terminate for async tasks):

     $eventDispatcher->addListener('kernel.terminate', function () {
         $this->get('old_sound_rabbit_mq.producer')->publish($data);
     });
     

2. Doctrine Integration:

   • Publish messages post-save (e.g., for notifications):

     $entityManager->getEventManager()->addEventListener(
         [PreUpdate::class, PostUpdate::class],
         new RabbitMqEventListener($this->get('old_sound_rabbit_mq.producer'))
     );
     

3. Retry Logic:

   • Implement exponential backoff for failed consumers:

     $attempts = 0;
     while ($attempts < 3) {
         try {
             $this->execute($message);
             break;
         } catch (Exception $e) {
             $attempts++;
             sleep(2 ** $attempts);
         }
     }
     

4. Monitoring:

   • Log message flow (e.g., monolog):

     # config/packages/monolog.yaml
     handlers:
         rabbitmq:
             type: stream
             path: "%kernel.logs_dir%/rabbitmq.log"
             level: debug
     

   • Use rabbitmq:status command to check connections:

     php bin/console rabbitmq:status
     

Gotchas and Tips

Pitfalls

1. Connection Management:

   • Lazy Connections: Set lazy: true in config to avoid connection overhead on demand.

     connections:
         default:
             lazy: true
     

   • Connection Drops: RabbitMQ may drop idle connections. Use heartbeat in config:

     connections:
         default:
             heartbeat: 30
     

2. Message Ordering:

   • RabbitMQ does not guarantee message order by default. Use priority in queue options if needed:

     queue_options:
         name: 'priority_queue'
         flags: {x-max-priority: 10}
     

3. Consumer Stuck:

   • If a consumer hangs, check for:
     • Infinite loops in callbacks.
     • Unhandled exceptions (wrap callbacks in try-catch).
     • Use --timeout in CLI to auto-stop:

       php bin/console rabbitmq:consumer consumer_name --timeout=60
       

4. Serialization Issues:

   • Avoid serializing closures/Resources. Use JSON or custom formats:

     $producer->publish(json_encode($data), 'routing_key');
     

     In consumer:

     $data = json_decode($message->body, true);
     

5. Permissions:

   • Ensure the RabbitMQ user has permissions for the vhost/exchange/queue:

     rabbitmqctl set_permissions -p / guest ".*" ".*" ".*"
     

Debugging

1. Check Queue Status:

   rabbitmqctl list_queues
   rabbitmqctl list_exchanges
   

2. Inspect Messages:

   • Use rabbitmq:consumer with --limit=1 to peek:

     php bin/console rabbitmq:consumer consumer_name --limit=1
     

   • Manually inspect queues with rabbitmqctl get_queue <queue_name>.

3. Enable Debugging:

   • Set debug: true in config to log connection details:

     old_sound_rabbit_mq:
         debug: true
     

4. Common Errors:

   • ConnectionException: Verify RabbitMQ server is running and credentials are correct.
   • ChannelException: Check for misconfigured exchanges/queues (e.g., typo in names).
   • AMQPTimeoutException: Increase timeout in config or check network latency.

Extension Points

1. Custom Producers/Consumers:

   • Extend ProducerInterface/ConsumerInterface for reusable logic:

     class AsyncEmailProducer implements ProducerInterface
     {
         public function publish($msg, $routingKey = '')
         {
             $this->get('old_sound_rabbit_mq.email_producer')->publish($msg, $routingKey);
         }
     }
     

2. Dynamic Exchanges/Queues:

   • Create exchanges/queues programmatically:

     $channel = $this->get('old_sound_rabbit_mq.default_producer_connection')->channel();
     $channel->exchange_declare('dynamic_exchange', 'direct', false, false, false);
     $channel->queue_declare('dynamic_queue', false, false, false, false);
     $channel->queue_bind('dynamic_queue', 'dynamic_exchange', 'routing_key');
     

3. Middleware for Messages:

   • Add preprocessing/postprocessing (e.g., logging, validation):

     $producer->publish(
         json_encode(['data' => $data, 'timestamp' => time()]),
         'routing_key'
     );
     

4. Testing:

   • Use `php