Phpamqplib Messenger Laravel Package

Getting Started

Minimal Setup

1. Install the package:

   composer require checkthiscloud/phpamqplib-messenger
   

2. Enable the bundle in config/bundles.php:

   Jwage\PhpAmqpLibMessengerBundle\PhpAmqpLibMessengerBundle::class => ['all' => true],
   

3. Configure a transport in config/messenger.php:

   'transports' => [
       'amqp' => [
           'dsn' => 'phpamqplib://guest:guest@localhost:5672/my_vhost/queue_name',
           'options' => [
               'exchange' => [
                   'name' => 'my_exchange',
                   'type' => 'direct',
               ],
               'queue' => [
                   'name' => 'my_queue',
                   'durable' => true,
               ],
           ],
       ],
   ],
   

4. Dispatch a test message:

   use Symfony\Component\Messenger\MessageBusInterface;
   
   $bus = app(MessageBusInterface::class);
   $bus->dispatch(new YourMessageClass());
   

First Use Case: Replacing php-amqp Extension

If your Laravel app currently uses the php-amqp extension (e.g., via laravel/amqp-messenger), migrate to this package by:

• Removing php-amqp from composer.json.
• Updating messenger.php to use the new DSN format (phpamqplib://).
• Testing consumers with php artisan messenger:consume amqp --time-limit=60.

Implementation Patterns

Core Workflows

1. Message Production

• Standard Dispatch:

  dispatch(new OrderProcessed($orderId)); // Uses default transport
  

• Explicit Transport:

  $bus->dispatch(new OrderProcessed($orderId), ['transport' => 'amqp']);
  

• Delayed Messages:

  $bus->dispatch(
      new SendEmail($userId),
      ['delay' => 3600] // Delay 1 hour
  );
  

2. Message Consumption

• Single Consumer:

  php artisan messenger:consume amqp --limit=10
  

• Worker Pool (for parallel processing):

  php artisan messenger:consume amqp --workers=4 --time-limit=300
  

• Signal Handling (for graceful shutdowns):

  // In a custom worker script (e.g., `artisan` command)
  declare(ticks=1);
  pcntl_signal(SIGTERM, function () {
      // Trigger shutdown logic
  });
  

3. Transport Configuration

• Advanced Options:

  # config/messenger.php
  transports:
      amqp:
          dsn: 'phpamqplib://user:pass@rabbitmq:5672/vhost?heartbeat=60'
          options:
              exchange:
                  name: 'orders.exchange'
                  type: 'topic'
                  durable: true
              queue:
                  name: 'orders.queue'
                  durable: true
                  arguments:
                      x-message-ttl: 86400000 # 24h TTL
  

• Multiple Transports:

  // Route messages to different transports
  $bus->dispatch(new HighPriorityMessage(), ['transport' => 'amqp_high']);
  

4. Middleware Integration

• Global Middleware:

  // app/Providers/AppServiceProvider.php
  public function boot()
  {
      $this->app->make(MessageBus::class)->add(
          new RetryFailedJobsMiddleware()
      );
  }
  

• Transport-Specific Middleware:

  # config/messenger.php
  transports:
      amqp:
          middleware:
              - 'doctrine_transaction'
              - 'amqp_confirmation' # Built-in for message acknowledgments
  

5. Connection Management

• Reusing Connections: The package manages connections efficiently. Avoid manual connection handling unless extending functionality (e.g., custom connection factories).
• Connection Retries: Enable automatic retries for transient failures:

  transports:
      amqp:
          retry_strategy:
              max_retries: 3
              delay: 1000
  

Gotchas and Tips

Pitfalls

1. DSN Format:

   • Gotcha: The DSN must use phpamqplib:// (not amqp:// or amqps://).
   • Fix: Update all transport configurations to match the new scheme.
   • Example:

     # Wrong
     dsn: 'amqp://user:pass@host:5672/vhost'
     # Right
     dsn: 'phpamqplib://user:pass@host:5672/vhost'
     

2. Symfony DI Dependency:

   • Gotcha: The package introduces symfony/dependency-injection, which may conflict with existing Symfony packages or Laravel’s native DI.
   • Fix:
     • Run composer why symfony/dependency-injection to check for conflicts.
     • Pin versions in composer.json if needed:

       "symfony/dependency-injection": "^6.0"
       

   • Tip: If using Laravel’s native Messenger, this dependency is unlikely to cause issues unless you’re also using Symfony’s DI container directly.

3. Message Idempotency:

   • Gotcha: The package enables publish confirms by default, which may cause duplicate messages if retries occur.
   • Fix: Ensure your message handlers are idempotent. Use unique message IDs or deduplication logic:

     class OrderProcessed implements MessageInterface
     {
         public function __construct(
             public string $orderId,
             public string $messageId // Ensure uniqueness
         ) {}
     }
     

4. Consumer Shutdown:

   • Gotcha: Unlike php-amqp, php-amqplib supports asynchronous consumers, but improper shutdowns (e.g., SIGKILL) can leave messages unacknowledged.
   • Fix: Use pcntl signals for graceful shutdowns:

     declare(ticks=1);
     pcntl_signal(SIGTERM, function () {
         // Trigger shutdown in your consumer
         $this->connection->close();
         exit(0);
     });
     

5. Queue Binding:

   • Gotcha: The package auto-creates exchanges and queues based on the DSN. Misconfigured bindings (e.g., wrong routing keys) may lead to silent message drops.
   • Fix: Explicitly define bindings in configuration:

     transports:
         amqp:
             options:
                 bindings:
                     - queue: 'orders.queue'
                       exchange: 'orders.exchange'
                       routing_key: 'order.created'
     

6. SSL/TLS Configuration:

   • Gotcha: SSL/TLS settings are not exposed in the default DSN. Custom connection factories may be needed for advanced setups.
   • Fix: Extend the transport configuration:

     // config/messenger.php
     transports:
         amqp_ssl:
             dsn: 'phpamqplib://user:pass@rabbitmq:5671/vhost'
             options:
                 ssl_options:
                     local_cert: '/path/to/cert.pem'
                     local_key: '/path/to/key.pem'
                     verify_peer: true
     

Debugging Tips

1. Enable Verbose Logging:

   php artisan messenger:consume amqp --verbose
   

   Or configure Monolog in config/logging.php:

   'channels' => [
       'messenger' => [
           'driver' => 'single',
           'path' => storage_path('logs/messenger.log'),
           'level' => 'debug',
       ],
   ],
   

2. Check Connection Health:

   • Use php-amqplib's built-in tools to verify connectivity:

     php -r "require 'vendor/autoload.php'; $conn = new PhpAmqpLib\Connection\AMQPStreamConnection('localhost', 5672, 'guest', 'guest'); echo 'Connected!';"
     

3. Monitor Failed Jobs:

   • Laravel’s failed_jobs table will log failed messages. Query it with:

     \Illuminate\Support\Facades\DB::table('failed_jobs')->where('connection', 'amqp')->get();
     

4. Inspect RabbitMQ:

   • Use the RabbitMQ management plugin (http://localhost:15672) to check:
     • Queue lengths.
     • Unacknowledged messages.
     • Consumer activity.

Extension Points

1. Custom Connection Factories:
   • Override the default connection logic by binding a custom factory:

     $this->app->bind(
         \