Messenger Enqueue Transport Laravel Package

Getting Started

Minimal Setup

1. Install the package:

   composer require autoprotect-group/messenger-enqueue-transport
   

2. Configure Enqueue Bundle (e.g., for RabbitMQ):

   # .env
   ENQUEUE_DSN=amqp://guest:guest@localhost:5672/%2f
   

3. Define Messenger Transport in config/packages/messenger.yaml:

   framework:
       messenger:
           transports:
               amqp: enqueue://default
           routing:
               'App\Message\MyMessage': amqp
   

4. Consume messages:

   php bin/console messenger:consume amqp
   

First Use Case

Send a message via the amqp transport:

use App\Message\MyMessage;
use Symfony\Component\Messenger\MessageBusInterface;

class MyController {
    public function __construct(private MessageBusInterface $bus) {}

    public function sendMessage() {
        $this->bus->dispatch(new MyMessage());
    }
}

Implementation Patterns

Core Workflow

1. Define Message Classes: Extend Symfony\Component\Messenger\Message or implement JsonSerializable.

   class MyMessage implements Message {
       public string $content;
   }
   

2. Route Messages:

   # config/packages/messenger.yaml
   framework:
       messenger:
           routing:
               'App\Message\MyMessage': amqp
   

3. Consume Messages:

   # Single worker
   php bin/console messenger:consume amqp --time-limit=300
   
   # Multiple workers (e.g., 5)
   php bin/console messenger:consume amqp --workers=5
   

4. Handle Messages:

   use App\Message\MyMessage;
   use Symfony\Component\Messenger\Attribute\AsMessageHandler;
   
   #[AsMessageHandler]
   public function handle(MyMessage $message) {
       // Process $message->content
   }
   

Advanced Patterns

1. Dynamic Transport Configuration: Use TransportConfiguration for runtime overrides:

   use Enqueue\MessengerAdapter\EnvelopeItem\TransportConfiguration;
   
   $transportConfig = (new TransportConfiguration())
       ->setDeliveryDelay(5000)
       ->addMetadata('deliveryMode', \Enqueue\AmqpExt\AmqpMessage::DELIVERY_MODE_PERSISTENT);
   
   $this->bus->dispatch((new Envelope($message))->with($transportConfig));
   

2. Topic-Based Routing:

   # config/packages/messenger.yaml
   transports:
       amqp:
           dsn: enqueue://default?queue[name]=my_queue&topic[name]=my_topic&topic[type]=topic
   

3. Retry Logic: Configure retry middleware in messenger.yaml:

   framework:
       messenger:
           failure_transport: failed
           transports:
               failed: 'doctrine://default?queue_name=failed'
           retry_strategy:
               max_retries: 3
               delay: 1000
   

4. Async Processing: Dispatch messages without waiting:

   $this->bus->dispatch(new MyMessage());
   // Continue execution...
   

Gotchas and Tips

Common Pitfalls

1. DSN Configuration:

   • Ensure ENQUEUE_DSN matches your broker (e.g., amqp://, redis://).
   • Gotcha: Missing ?queue[name] in DSN may default to default queue, causing collisions.

2. Message Serialization:

   • Gotcha: Non-serializable messages (e.g., closures, resources) will fail silently. Use JsonSerializable or @Serializer\SerializedName.
   • Fix: Add @Serializer\Ignore to problematic properties.

3. Worker Lifecycle:

   • Gotcha: Workers exit on unhandled exceptions. Use --limit or --time-limit to avoid abrupt stops.
   • Tip: Log worker PID for debugging:

     php bin/console messenger:consume amqp --limit=100 --logger-format="[%pid%] %message%"
     

4. Transport-Specific Quirks:

   • AMQP: Persistent messages require deliveryMode: 2 (see TransportConfiguration).
   • Redis: Use ?queue[serializer]=json if messages fail to deserialize.
   • SQS: Ensure ENQUEUE_DSN includes ?queue[url]=... for FIFO queues.

5. Symfony Version Mismatches:

   • Gotcha: Package supports Symfony 4.3+, 5.x, and 6.x. Mixing versions (e.g., symfony/messenger:^4.3 with Symfony 6) may cause autowiring issues.
   • Fix: Align dependencies in composer.json.

Debugging Tips

1. Check Queue State:

   # RabbitMQ
   rabbitmqctl list_queues
   
   # Redis
   redis-cli LRANGE my_queue 0 -1
   

2. Enable Messenger Debug:

   # config/packages/dev/messenger.yaml
   framework:
       messenger:
           transports:
               amqp: '%env(MESSENGER_TRANSPORT_DSN)%'
           bus:
               middleware:
                   - Symfony\Component\Messenger\Middleware\HandleMessageMiddleware
                   - Symfony\Component\Messenger\Middleware\SendMessageMiddleware
                   - Symfony\Component\Messenger\Middleware\Stamp\StampMiddleware
   

3. Log Transport Configuration: Override Enqueue\MessengerAdapter\Transport\Transport to log DSN parsing:

   public function __construct(string $dsn) {
       parent::__construct($dsn);
       \Log::debug('Transport DSN:', ['dsn' => $dsn]);
   }
   

Extension Points

1. Custom Message Classes: Extend Enqueue\MessengerAdapter\Message\MessageInterface for broker-specific features:

   class CustomAmqpMessage extends \Enqueue\AmqpExt\AmqpMessage {
       public function setCustomHeader(string $key, $value) {
           $this->getProperties()->setHeader($key, $value);
       }
   }
   

2. Middleware Integration: Add transport-specific logic via TransportConfiguration:

   $transportConfig = new TransportConfiguration([
       'metadata' => [
           'customHeader' => 'value',
       ],
   ]);
   

3. Event Listeners: Subscribe to MessengerMessageToTransport events to inspect/modify messages:

   use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
   
   public function onWorkerStopped(WorkerStoppedEvent $event) {
       if ($event->getWorker()->getTransportName() === 'amqp') {
           \Log::info('AMQP worker stopped', ['reason' => $event->getReason()]);
       }
   }
   

4. Testing: Use Enqueue\Test\Mock\MockContext for unit tests:

   use Enqueue\Test\Mock\MockContext;
   
   $context = new MockContext();
   $transport = new \Enqueue\MessengerAdapter\Transport\Transport('enqueue://default', $context);