Amqp Bunny Laravel Package

Getting Started

Minimal Setup in Laravel

1. Install the package:

   composer require enqueue/amqp-bunny
   

   Ensure bunny/bunny (AMQP client) and queue-interop/amqp-interop are also installed.

2. Configure the transport: Add to config/queue.php under connections:

   'amqp-bunny' => [
       'dsn' => env('AMQP_DSN', 'amqp://guest:guest@localhost:5672/%2f'),
       'options' => [
           'connection_timeout' => 3.0,
           'read_write_timeout' => 3.0,
       ],
       'queue' => env('AMQP_QUEUE', 'default'),
       'exchange' => env('AMQP_EXCHANGE', 'default'),
       'routing_key' => env('AMQP_ROUTING_KEY', 'default'),
       'prefetch_count' => env('AMQP_PREFETCH', 1),
   ],
   

3. First use case: Push a job to the queue:

   use Enqueue\AmqpBunny\AmqpBunnyConnectionFactory;
   use Enqueue\Client\Producer;
   
   $connectionFactory = new AmqpBunnyConnectionFactory();
   $connection = $connectionFactory->createConnection(['dsn' => config('queue.connections.amqp-bunny.dsn')]);
   $producer = new Producer($connection);
   $producer->send(new Message('Hello, AMQP!'));
   

Implementation Patterns

Workflows

1. Producer-Consumer Pattern:

   • Producer: Dispatch jobs to AMQP:

     $producer = $connection->createProducer();
     $producer->send(new Message('Job payload', ['priority' => 5]));
     

   • Consumer: Process messages in a loop:

     $consumer = $connection->createConsumer(config('queue.connections.amqp-bunny.queue'));
     while ($message = $consumer->receive()) {
         try {
             // Process message
             $message->ack();
         } catch (\Exception $e) {
             $message->reject();
         }
     }
     

2. Laravel Queue Integration: Extend Laravel’s queue system to use AMQP:

   // config/queue.php
   'default' => env('QUEUE_CONNECTION', 'amqp-bunny'),
   

   Dispatch jobs as usual:

   dispatch(new ProcessPodcast());
   

3. Error Handling: Use Message::reject() with a requeue flag for transient failures:

   if ($message->getPayload() === 'fail') {
       $message->reject(false); // Do not requeue
   }
   

Integration Tips

• Laravel Horizon: Configure Horizon to use AMQP by extending HorizonServiceProvider and overriding connection().
• Retry Logic: Implement a dead-letter queue (DLQ) for failed messages:

  $consumer = $connection->createConsumer('dlq');
  $consumer->setMessageHandler(function (Message $message) {
      // Log and reprocess
  });
  

• Connection Management: Reuse connections for performance:

  $connection = $connectionFactory->createConnection(['dsn' => $dsn]);
  $producer = new Producer($connection);
  

Gotchas and Tips

Pitfalls

1. Connection Timeouts:

   • Bunny’s default timeouts (3s) may be too short for slow networks. Increase read_write_timeout in config.
   • Fix: Add to config/queue.php:

     'options' => [
         'read_write_timeout' => 10.0,
     ],
     

2. Message Ordering:

   • AMQP does not guarantee FIFO by default. Use priority headers or a single-consumer queue for ordered processing.

3. Consumer Lag:

   • High prefetch_count can overwhelm workers. Start with 1 and adjust based on workload.

4. Queue Binding:

   • If using exchanges, ensure queues are bound to the correct exchange with the right routing_key:

     $connection->createChannel()->queueBind(
         config('queue.connections.amqp-bunny.queue'),
         config('queue.connections.amqp-bunny.exchange'),
         config('queue.connections.amqp-bunny.routing_key')
     );
     

Debugging

• Enable Bunny Debugging: Set the BUNNY_DEBUG environment variable to log raw AMQP commands:

  export BUNNY_DEBUG=1
  

• Check Connection Health: Use bunny/bunny's Connection::isConnected() to verify connectivity before producing/consuming.

Extension Points

1. Custom Message Serialization: Override the default JSON serializer:

   $producer->setSerializer(new \Enqueue\AmqpBunny\Serializer\JsonSerializer());
   

   Or implement MessageSerializerInterface for custom formats (e.g., Protobuf).

2. Middleware: Add processing logic before/after message handling:

   $consumer->setMiddleware([
       new \Enqueue\AmqpBunny\Middleware\LoggingMiddleware(),
       new class implements Middleware {
           public function handle(Message $message, callable $next) {
               // Pre-processing
               $result = $next($message);
               // Post-processing
               return $result;
           }
       },
   ]);
   

3. Dynamic Queue Routing: Use Message::setProperty() to dynamically route messages:

   $message->setProperty('x-priority', 10); // High priority
   $producer->send($message);