Laravel Rabbitmq Laravel Package

## Getting Started

### Minimal Setup
1. **Installation**
   ```bash
   composer require iamfarhad/laravel-rabbitmq:^1.2.0

Publish the config (optional but recommended for customization):

php artisan vendor:publish --provider="Farhad\RabbitMQ\RabbitMQServiceProvider" --tag="config"

2. Configure .env

   QUEUE_CONNECTION=rabbitmq
   RABBITMQ_HOST=localhost
   RABBITMQ_PORT=5672
   RABBITMQ_USER=guest
   RABBITMQ_PASSWORD=guest
   

3. First Use Case: Dispatch a Job

   use App\Jobs\ProcessOrder;
   
   ProcessOrder::dispatch($orderId); // Uses RabbitMQ as the default queue
   

4. Verify Connection Check the rabbitmq:monitor command for connection health:

   php artisan rabbitmq:monitor
   

Implementation Patterns

Core Workflows

1. Queue Configuration

• Default Queue: Uses queue as the default exchange/queue name (configurable in config/rabbitmq.php).
• Custom Queues: Define queues in config/rabbitmq.php under queues:

  'queues' => [
      'orders' => [
          'exchange' => 'orders_exchange',
          'queue' => 'orders_queue',
          'binding_key' => 'order.*',
          'durable' => true,
          'dead_letter_exchange' => 'orders_dlq',
          'dead_letter_routing_key' => 'dlq.order',
      ],
  ],
  

• Dynamic Queues: Create queues programmatically:

  $this->rabbitMQ->queue('dynamic_queue', ['durable' => true]);
  

2. Job Dispatching

• Basic Dispatch:

  ProcessOrder::dispatch($orderId)->onQueue('orders');
  

• Priority Queues: Use priority option (requires RabbitMQ priority plugin):

  ProcessOrder::dispatch($orderId)->onQueue('high_priority')->priority(1);
  

• Delayed Jobs: Leverage RabbitMQ’s built-in delayed message support:

  ProcessOrder::dispatch($orderId)->delay(now()->addMinutes(10));
  

3. Consumer Patterns

• Single Consumer:

  php artisan queue:work rabbitmq --queue=orders
  

• Daemon Mode (for production):

  php artisan queue:work rabbitmq --daemon --sleep=3 --tries=3
  

• Custom Consumer Logic:

  $this->rabbitMQ->consume('orders', function ($job) {
      // Custom processing logic
      $job->delete(); // Manually acknowledge
  });
  

4. Connection Management

• Explicit Connection Handling:

  $connection = $this->rabbitMQ->connection();
  $channel = $connection->channel();
  

• Connection Pooling: The package auto-manages a pool of connections/channels (configurable via config/rabbitmq.php):

  'pool' => [
      'connections' => 5,
      'channels' => 10,
  ],
  

5. Pub/Sub Model

• Publish Messages:

  $this->rabbitMQ->publish('order.created', ['order_id' => 123]);
  

• Subscribe to Events:

  $this->rabbitMQ->subscribe('order.*', function ($message) {
      // Handle message
  });
  

6. Error Handling & Retries

• Retryable Jobs: Use Laravel’s built-in retry mechanism:

  ProcessOrder::dispatch($orderId)->retryAfter(5);
  

• Custom Retry Logic:

  ProcessOrder::dispatch($orderId)->retryUntil(function () {
      return DB::table('orders')->where('id', $orderId)->exists();
  });
  

Integration Tips

Laravel Ecosystem

• Queue Workers: Replace database/sync drivers with rabbitmq in QUEUE_CONNECTION.
• Horizon: Works seamlessly with Laravel Horizon for monitoring.
• Events: Dispatch events to RabbitMQ queues:

  event(new OrderCreated($orderId))->onQueue('events');
  

Performance Optimization

• Batch Processing: Use chunk() for large datasets:

  User::chunk(100, function ($users) {
      ProcessUsers::dispatch($users)->onQueue('batch_users');
  });
  

• Async Processing: Offload CPU-intensive tasks:

  GenerateReport::dispatch($reportId)->onQueue('reports');
  

Monitoring

• Built-in Metrics:

  $stats = $this->rabbitMQ->stats();
  // Access: $stats->connections, $stats->channels, $stats->messages
  

• Prometheus Integration: Expose metrics via rabbitmq:monitor command or custom middleware.

Gotchas and Tips

Pitfalls

1. Connection Leaks

   • Issue: Unclosed channels/connections can exhaust RabbitMQ limits.
   • Fix: Ensure try-catch blocks close resources:

     try {
         $channel = $this->rabbitMQ->channel();
         // Work with channel
     } finally {
         $channel->close();
     }
     

   • Tip: Use RabbitMQ::withChannel() for scoped channel management:

     $this->rabbitMQ->withChannel(function ($channel) {
         // Channel auto-closes after block
     });
     

2. Message Loss

   • Issue: Unacknowledged messages may be redelivered or lost.
   • Fix: Always acknowledge messages in consumers:

     $this->rabbitMQ->consume('queue', function ($job) {
         try {
             // Process job
             $job->ack(); // Explicit acknowledge
         } catch (\Exception $e) {
             $job->nack(); // Negative acknowledge (requeue)
         }
     });
     

   • Tip: Use auto_ack: false in queue config for manual control.

3. Exponential Backoff Misconfiguration

   • Issue: Retries may fail if backoff is too aggressive.
   • Fix: Configure retry strategy in config/rabbitmq.php:

     'retry' => [
         'max_attempts' => 5,
         'backoff' => [1, 2, 3, 5, 8], // Custom delays in seconds
     ],
     

4. Queue Binding Errors

   • Issue: Incorrect binding_key or missing exchanges can cause silent failures.
   • Fix: Verify bindings with:

     php artisan rabbitmq:monitor
     

   • Tip: Use declareQueue() to ensure queues/exchanges exist:

     $this->rabbitMQ->declareQueue('orders', ['durable' => true]);
     

5. Memory Leaks in Consumers

   • Issue: Long-running consumers may hold channels open.
   • Fix: Use sleep() between batches or limit consumer runtime:

     php artisan queue:work rabbitmq --sleep=5 --memory=128
     

6. Compatibility Issues

   • Issue: Potential compatibility bugs fixed in 1.2.0 may affect older Laravel versions.
   • Fix: Ensure compatibility by updating Laravel and dependencies:

     composer update laravel/framework --with-all-dependencies
     

   • Tip: Test thoroughly after upgrading, especially if using custom queue configurations.

Debugging

1. Enable Debug Logging Add to .env:

   RABBITMQ_LOG_LEVEL=debug
   

   View logs with:

   tail -f storage/logs/laravel.log
   

2. RabbitMQ Management UI Access http://localhost:15672 (default credentials: guest/guest) to inspect queues, messages, and connections.

3. Common Errors

   • AMQPConnectionException: Check credentials, network, or RabbitMQ service status.
   • ChannelException: Likely a channel leak or misconfiguration. Run rabbitmq:monitor to check pool stats.
   • NotFoundException: Queue/exchange not declared. Use declareQueue() or verify bindings.

Tips

1. Configuration Overrides Override settings per environment:

   config(['rabbitmq.queues.orders.durable' => env('RABBITMQ_ORDERS_DURABLE', true)]);
   

2. Dynamic Queue Routing Route jobs dynamically based on payload:

   ProcessOrder::dispatch($orderId)