Rabbitmqbundle Laravel Package

Getting Started

Minimal Setup

1. Installation Add the package via Composer:

   composer require xlabs/rabbitmqbundle
   

   Register the bundle in config/app.php under providers:

   Xlabs\RabbitMQBundle\RabbitMQServiceProvider::class,
   

2. Configuration Publish the default config:

   php artisan vendor:publish --provider="Xlabs\RabbitMQBundle\RabbitMQServiceProvider" --tag=config
   

   Update config/rabbitmq.php with your RabbitMQ server details (host, port, credentials, etc.).

3. First Use Case: Publishing a Message Inject the RabbitMQConnection service and publish a message:

   use Xlabs\RabbitMQBundle\RabbitMQConnection;
   
   public function __construct(RabbitMQConnection $rabbitMQ)
   {
       $this->rabbitMQ = $rabbitMQ;
   }
   
   public function sendMessage()
   {
       $this->rabbitMQ->publish(
           'exchange_name',
           'routing_key',
           'message_body',
           ['content_type' => 'application/json']
       );
   }
   

4. First Use Case: Consuming a Message Define a consumer class and register it in config/rabbitmq.php under consumers:

   'consumers' => [
       'my_consumer' => [
           'queue' => 'my_queue',
           'class' => \App\Consumers\MyConsumer::class,
           'callback' => 'handleMessage',
       ],
   ],
   

   Implement the consumer:

   namespace App\Consumers;
   
   class MyConsumer
   {
       public function handleMessage($message)
       {
           // Process message
           return $message; // Acknowledge
       }
   }
   

Implementation Patterns

Workflows

1. Producer Workflow

   • Publish Messages: Use publish() for direct exchanges or queuePublish() for queues.
   • Delayed Messages: Leverage RabbitMQ’s built-in TTL or use a dead-letter exchange.

     $this->rabbitMQ->publish('exchange', 'routing_key', $message, [
         'content_type' => 'application/json',
         'expiration' => 5000, // 5 seconds in ms
     ]);
     

   • Batch Processing: Publish multiple messages in a loop or use transactions for atomicity.

     $this->rabbitMQ->txSelect();
     foreach ($messages as $message) {
         $this->rabbitMQ->publish('exchange', 'key', $message);
     }
     $this->rabbitMQ->txCommit();
     

2. Consumer Workflow

   • Declare Consumers: Register consumers in config/rabbitmq.php with their queues, classes, and methods.
   • Prefetch Count: Control concurrency by setting prefetch_count in the config.
   • Error Handling: Use try-catch in consumer methods and log failures.

     public function handleMessage($message)
     {
         try {
             // Process logic
         } catch (\Exception $e) {
             \Log::error('Consumer error: ' . $e->getMessage());
             throw $e; // Reject message
         }
     }
     

   • Dynamic Consumers: For dynamic queues, use RabbitMQConnection directly to declare queues/exchanges and bind consumers at runtime.

3. Integration with Laravel Jobs

   • Dispatch jobs from consumers or publish messages from jobs:

     // Inside consumer
     dispatch(new ProcessOrder($orderData));
     
     // Inside job
     $this->rabbitMQ->publish('orders', 'new_order', $orderData);
     

Patterns

1. Exchange and Queue Management

   • Declare exchanges/queues programmatically:

     $this->rabbitMQ->declareExchange('my_exchange', 'direct', false, false, false);
     $this->rabbitMQ->declareQueue('my_queue', false, false, false);
     $this->rabbitMQ->bindQueue('my_queue', 'my_exchange', 'routing.key');
     

2. Message Serialization

   • Customize serialization in config:

     'serializer' => [
         'class' => \App\Services\CustomSerializer::class,
         'method' => 'serialize',
     ],
     

   • Implement CustomSerializer:

     class CustomSerializer
     {
         public function serialize($data) { /* ... */ }
         public function unserialize($data) { /* ... */ }
     }
     

3. Retry Mechanism

   • Use RabbitMQ’s requeue or implement a retry queue:

     public function handleMessage($message)
     {
         if ($failed) {
             $this->rabbitMQ->reject($message, false); // Requeue
         }
     }
     

4. Monitoring and Metrics

   • Log message flow or integrate with Laravel’s logging:

     \Log::info('Message published', ['exchange' => 'exchange_name', 'routing_key' => 'key']);
     

Gotchas and Tips

Pitfalls

1. Connection Management

   • Issue: Connection drops or timeouts during long-running consumers.
   • Fix: Use reconnect() or implement a heartbeat in the config:

     'connection' => [
         'heartbeat' => 60, // seconds
     ],
     

   • Tip: Wrap consumer logic in a loop with reconnection logic:

     while (true) {
         try {
             $this->rabbitMQ->consume();
         } catch (\Exception $e) {
             sleep(5);
             $this->rabbitMQ->reconnect();
         }
     }
     

2. Message Acknowledgment

   • Issue: Unacknowledged messages pile up if consumers crash.
   • Fix: Use auto_ack: false in config and manually acknowledge:

     'consumers' => [
         'my_consumer' => [
             'auto_ack' => false,
             // ...
         ],
     ],
     

   • Tip: Always return the message in the consumer method to auto-acknowledge:

     public function handleMessage($message) {
         // Process
         return $message; // Acknowledges
     }
     

3. Queue Binding

   • Issue: Messages not reaching queues due to incorrect bindings.
   • Fix: Verify bindings with:

     $this->rabbitMQ->bindQueue('queue', 'exchange', 'routing.key');
     

   • Tip: Use declareQueue() with passive: true to check if a queue exists:

     $this->rabbitMQ->declareQueue('queue', true, false, false, false);
     

4. Configuration Overrides

   • Issue: Config changes not reflected due to caching.
   • Fix: Clear config cache:

     php artisan config:clear
     

   • Tip: Use environment variables for dynamic config:

     'host' => env('RABBITMQ_HOST', 'localhost'),
     

Debugging

1. Enable Debugging

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

     'debug' => env('RABBITMQ_DEBUG', false),
     

2. Check Connection

   • Test connectivity with:

     $this->rabbitMQ->isConnected();
     

3. Inspect Queues/Exchanges

   • List queues/exchanges:

     $queues = $this->rabbitMQ->listQueues();
     $exchanges = $this->rabbitMQ->listExchanges();
     

4. Message Inspection

   • Log raw messages for debugging:

     \Log::debug('Raw message', ['body' => $message->body]);
     

Extension Points

1. Custom Middleware

   • Add middleware for pre/post-processing:

     'middleware' => [
         \App\Middleware\LoggingMiddleware::class,
     ],
     

   • Implement middleware:

     class LoggingMiddleware
     {
         public function handle($message, \Closure $next)
         {
             \Log::info('Message processed', ['body' => $message->body]);
             return $next($message);
         }
     }
     

2. Event Dispatching

   • Trigger Laravel events from consumers:

     event(new OrderProcessed($orderData));
     

3. Dynamic Queue Declaration

   • Declare queues dynamically based on runtime conditions:

     $queueName = 'dynamic_queue_' . $userId;
     $this->rabbitMQ->declareQueue($queueName, false, false, false, ['exclusive' => true]);
     

4. Plugin System

   • Extend functionality by creating plugins (e.g., for monitoring or metrics):

     $this->rabbitMQ->plugin('monitoring')->track('event_name');