Amqp Ext Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require enqueue/amqp-ext
   

   Ensure ext-amqp is enabled in your PHP environment (php -m | grep amqp).

2. Basic Usage:

   use Enqueue\AmqpExt\AmqpConnectionFactory;
   use Enqueue\AmqpExt\AmqpContext;
   
   $connectionFactory = new AmqpConnectionFactory('amqp://guest:guest@localhost');
   $context = new AmqpContext($connectionFactory);
   $producer = $context->createProducer();
   $producer->send(new \Enqueue\Message\Message('Hello, AMQP!'), ['queue' => 'test']);
   

3. First Use Case:

   • Send a delayed message:

     $producer->send(new \Enqueue\Message\Message('Delayed'), [
         'queue' => 'delayed',
         'delay' => 1000 // 1 second
     ]);
     

Where to Look First

• Documentation for transport-specific features.
• AmqpConnectionFactory for connection configuration (host, credentials, SSL).
• AmqpContext for managing producers/consumers.

Implementation Patterns

Core Workflows

1. Producer-Consumer Pattern:

   // Producer
   $producer = $context->createProducer();
   $producer->send(new \Enqueue\Message\Message('Task'), ['queue' => 'tasks']);
   
   // Consumer (with manual acknowledgment)
   $consumer = $context->createConsumer('tasks');
   $consumer->consume(function (\Enqueue\Message\Message $message, \Enqueue\Context $context) {
       // Process message
       $context->ack($message);
   });
   

2. Connection Management:

   • Reuse AmqpContext for multiple producers/consumers.
   • Configure connection pooling via AmqpConnectionFactory:

     $factory = new AmqpConnectionFactory('amqp://user:pass@localhost', [
         'connection_options' => [
             'read_timeout' => 5.0,
             'write_timeout' => 5.0,
         ],
     ]);
     

3. Error Handling:

   • Wrap operations in try-catch for Enqueue\Exception\ExceptionInterface.
   • Use AmqpContext::createConsumer() with a callable for graceful error recovery.

Integration Tips

• Laravel Integration: Bind the context to Laravel’s service container:

  $app->singleton(AmqpContext::class, function ($app) {
      return new AmqpContext(new AmqpConnectionFactory(config('queue.amqp.url')));
  });
  

  Use Laravel’s queue system with enqueue/laravel:

  composer require enqueue/laravel
  

• Retry Logic: Leverage Enqueue\AmqpExt\AmqpConnectionFactory with retry_strategy:

  $factory->setRetryStrategy(new \Enqueue\Retry\ExponentialRetryStrategy());
  

• Monitoring: Enable AMQP metrics via amqp-tools:

  $context = new AmqpContext($factory, new \Enqueue\AmqpExt\AmqpTools());
  

Gotchas and Tips

Pitfalls

1. Connection Leaks:

   • Issue: Unclosed channels/connections can exhaust AMQP server resources.
   • Fix: Use AmqpContext as a singleton or ensure close() is called:

     $context->close();
     

2. Message Serialization:

   • Issue: AMQP messages must be serializable. Use Enqueue\Message\JsonMessage for complex payloads:

     $message = new \Enqueue\Message\JsonMessage(['data' => 'complex']);
     

3. Consumer Lag:

   • Issue: Slow consumers cause queue backlog.
   • Fix: Adjust prefetch count:

     $consumer = $context->createConsumer('queue', [
         'prefetch_count' => 10, // Default: 1
     ]);
     

4. SSL/TLS:

   • Issue: Self-signed certificates may fail silently.
   • Fix: Configure CA path explicitly:

     $factory = new AmqpConnectionFactory('amqp://user:pass@localhost', [
         'connection_options' => [
             'ssl_options' => [
                 'cafile' => '/path/to/ca.crt',
                 'verify_peer' => true,
             ],
         ],
     ]);
     

Debugging

• Enable AMQP Debugging:

  putenv('AMQP_DEBUG=1');
  

  Logs will appear in stderr.

• Check Connection Status:

  if (!$context->isConnected()) {
      $context->reconnect();
  }
  

Extension Points

1. Custom Message Handlers:

   • Extend Enqueue\AmqpExt\AmqpMessage for custom headers/serialization.

2. Middleware:

   • Use Enqueue\AmqpExt\Middleware\Middleware to intercept messages:

     $context = new AmqpContext($factory, [], [
         new class implements \Enqueue\Middleware\Middleware {
             public function handle(\Enqueue\Message\Message $message, callable $next) {
                 // Pre-processing
                 $result = $next($message);
                 // Post-processing
                 return $result;
             }
         },
     ]);
     

3. Plugin System:

   • Integrate with enqueue/amqp-tools for advanced features like:
     • Dead-letter exchanges.
     • Message TTL (Time-To-Live).

Config Quirks

• Default Exchange: The transport uses a direct exchange by default. For other exchange types (e.g., fanout), configure via AmqpContext:

  $context = new AmqpContext($factory, [], [
      'exchange' => [
          'name' => 'custom_exchange',
          'type' => 'fanout',
      ],
  ]);
  

• Queue Declarations: Ensure queues/exchanges are declared before use. Use AmqpTools for declarative setup:

  $tools = new \Enqueue\AmqpExt\AmqpTools($context);
  $tools->declareQueue('critical_tasks', ['durable' => true]);