Amqp Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation Add the bundle via Composer:

   composer require ecommit/amqp-bundle
   

   Enable the bundle in config/bundles.php:

   Ecommit\AmqpBundle\EcommitAmqpBundle::class => ['all' => true],
   

2. Configuration Publish the default config:

   php bin/console config:dump-reference EcommitAmqpBundle
   

   Configure config/packages/ecommit_amqp.yaml:

   ecommit_amqp:
       host: 'amqp://guest:guest@localhost:5672'
       exchange: 'your_exchange'
       queue: 'your_queue'
       routing_key: 'your.routing.key'
   

3. First Use Case Inject the AmqpProducer service in a controller or command:

   use Ecommit\AmqpBundle\Producer\AmqpProducerInterface;
   
   class ExampleController extends AbstractController
   {
       public function __construct(private AmqpProducerInterface $amqpProducer) {}
   
       public function sendMessage()
       {
           $this->amqpProducer->publish('Hello AMQP!');
           return new Response('Message sent!');
       }
   }
   

Implementation Patterns

Core Workflows

1. Message Production Use AmqpProducerInterface to publish messages:

   $this->amqpProducer->publish(
       json_encode(['event' => 'order.created', 'data' => $order]),
       ['content_type' => 'application/json']
   );
   

2. Message Consumption Create a consumer service implementing AmqpConsumerInterface:

   use Ecommit\AmqpBundle\Consumer\AmqpConsumerInterface;
   
   class OrderConsumer implements AmqpConsumerInterface
   {
       public function consume($message)
       {
           $data = json_decode($message, true);
           // Process $data['event'] and $data['data']
       }
   }
   

   Register the consumer in services.yaml:

   services:
       App\Consumer\OrderConsumer:
           tags:
               - { name: ecommit_amqp.consumer, queue: 'order_queue' }
   

3. Error Handling Implement AmqpErrorHandlerInterface for custom error logic:

   use Ecommit\AmqpBundle\Error\AmqpErrorHandlerInterface;
   
   class CustomErrorHandler implements AmqpErrorHandlerInterface
   {
       public function handle(\PhpAmqpLib\Exception\AMQPException $e)
       {
           // Log, retry, or notify
       }
   }
   

   Configure in config/packages/ecommit_amqp.yaml:

   ecommit_amqp:
       error_handler: 'App\Error\CustomErrorHandler'
   

Integration Tips

• Symfony Messenger Bridge Use AmqpTransport to integrate with Symfony Messenger:

  # config/packages/messenger.yaml
  messenger:
      transports:
          amqp:
              dsn: '%env(AMQP_DSN)%'
              options:
                  exchange: ['your_exchange', 'direct']
  

• Retry Logic Leverage AmqpRetryStrategy for transient failures:

  ecommit_amqp:
      retry_strategy:
          max_retries: 3
          delay: 1000
  

Gotchas and Tips

Pitfalls

1. Connection Management

   • Issue: Unclosed channels/connections can exhaust resources.
   • Fix: Use AmqpProducer::close() in shutdown events or implement __destruct() in consumers.

2. Message Serialization

   • Issue: Default publish() uses serialize(). Explicitly set content_type for JSON:

     $this->amqpProducer->publish(json_encode($data), ['content_type' => 'application/json']);
     

3. Queue Binding

   • Issue: Forgetting to bind queues to exchanges causes silent failures.
   • Fix: Verify bindings via php bin/console debug:container Ecommit\AmqpBundle\Manager\AmqpManager.

4. Consumer Lifecycle

   • Issue: Consumers may not auto-start if not tagged correctly.
   • Fix: Ensure eccommit_amqp.consumer tag includes queue and concurrency (default: 1).

Debugging

• Enable AMQP Debugging Add to config/packages/ecommit_amqp.yaml:

  ecommit_amqp:
      debug: true
  

  Logs will appear in var/log/dev.log.

• Check Queue Status Use php-amqplib CLI tools or RabbitMQ Management UI to verify:

  • Queue lengths.
  • Unacked messages.
  • Consumer tags.

Extension Points

1. Custom Serializers Override AmqpProducer to support custom formats:

   class CustomAmqpProducer extends AbstractAmqpProducer
   {
       protected function serialize($message, array $headers = [])
       {
           return json_encode($message);
       }
   }
   

   Register as a service with the amqp.producer tag.

2. Dynamic Routing Use AmqpProducer::publishWithRoutingKey() for dynamic keys:

   $this->amqpProducer->publishWithRoutingKey(
       $message,
       'dynamic.' . $userId . '.key'
   );
   

3. Middleware Support Extend AmqpProducer to add middleware (e.g., logging, validation):

   $producer = new AmqpProducer($connection, $exchange);
   $producer->addMiddleware(new LoggingMiddleware());