Amqp Lib Laravel Package

Getting Started

Minimal Setup in Laravel

1. Installation:

   composer require enqueue/amqp-lib enqueue/amqp-ext
   

   Ensure php-amqplib is installed (php-amqplib/php-amqplib).

2. Configure RabbitMQ Connection: Add to config/queue.php under connections:

   'amqp' => [
       'driver' => 'amqp',
       'host' => env('AMQP_HOST', 'localhost'),
       'port' => env('AMQP_PORT', 5672),
       'user' => env('AMQP_USER', 'guest'),
       'password' => env('AMQP_PASSWORD', 'guest'),
       'vhost' => env('AMQP_VHOST', '/'),
       'options' => [
           'ssl_options' => [
               'cafile' => env('AMQP_SSL_CAFILE', null),
           ],
       ],
   ],
   

3. First Use Case: Dispatch a job using the AMQP queue:

   use Illuminate\Support\Facades\Queue;
   
   Queue::connection('amqp')->push(new YourJobClass);
   

4. Verify Connection: Run a worker:

   php artisan queue:work --queue=default --connection=amqp
   

Implementation Patterns

Core Workflows

1. Job Dispatching: Use Laravel’s Queue::connection('amqp')->push() for AMQP-specific queues.

   Queue::connection('amqp')->push(
       new ProcessPodcast,
       'podcasts' // Custom queue name
   );
   

2. Consuming Messages: Extend Enqueue\AmqpLib\AmqpContext for custom AMQP logic:

   $context = new AmqpContext([
       'host' => 'localhost',
       'user' => 'guest',
       'pass' => 'guest',
   ]);
   $consumer = new AmqpConsumer($context, 'your_queue');
   $consumer->consume(function (Message $message) {
       // Process message
       $message->ack();
   });
   

3. Delayed Jobs: Use delay() with AMQP’s built-in scheduling:

   Queue::connection('amqp')->later(now()->addMinutes(10), new YourJob);
   

4. Fanout Exchanges: For pub/sub patterns, configure a fanout exchange:

   $context = new AmqpContext([...]);
   $exchange = new FanoutExchange($context, 'events');
   $exchange->publish(new AmqpMessage('data'));
   

Integration Tips

• Laravel Events: Dispatch events to AMQP queues:

  event(new PodcastProcessed($podcast))
      ->onQueue('podcast_events', 'amqp');
  

• Retry Logic: Leverage AMQP’s dead-letter exchanges (DLX) for failed jobs:

  'amqp' => [
      'driver' => 'amqp',
      'options' => [
          'dead_letter_exchange' => 'failed_jobs',
      ],
  ],
  

• Monitoring: Use amqp-tools for queue inspection:

  composer require enqueue/amqp-tools
  php vendor/bin/amqp-tools list-queues
  

Gotchas and Tips

Pitfalls

1. Connection Handling:

   • AMQP connections are not automatically reconnected on failure. Use a connection pool or retry logic:

     try {
         $context->createProducer()->send($message);
     } catch (AmqpConnectionException $e) {
         retry()->times(3)->then(function () use ($message) {
             $context->createProducer()->send($message);
         });
     }
     

   • Tip: Configure reconnect_delay in options for graceful retries.

2. Message Persistence:

   • Messages are not persisted by default. Enable durability:

     $queue = new Queue($context, 'your_queue', Queue::DURABLE);
     

3. Consumer Lag:

   • High consumer lag may block the queue. Use prefetch counts:

     $consumer->setPrefetchCount(10); // Process 10 messages at a time
     

4. SSL/TLS:

   • If using SSL, ensure ssl_options are correctly configured. Test with:

     openssl s_client -connect localhost:5671 -showcerts
     

Debugging

• Enable Logging:

  $context = new AmqpContext([...], [
      'logger' => new MonologLogger(new Logger('amqp')),
  ]);
  

• Check AMQP Server Logs: RabbitMQ logs are typically at /var/log/rabbitmq/rabbit@hostname.log.

Extension Points

1. Custom Serializers: Override the default JSON serializer:

   $context = new AmqpContext([...], [
       'serializer' => new YourCustomSerializer(),
   ]);
   

2. Middleware: Add AMQP-specific middleware to Laravel’s queue:

   Queue::connection('amqp')->middleware([
       new AmqpLoggingMiddleware(),
   ]);
   

3. Plugin System: Extend AmqpContext to add custom plugins (e.g., metrics):

   $context = new AmqpContext([...]);
   $context->addPlugin(new AmqpMetricsPlugin());