Messenger Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require alpari/messenger-bundle
   

   Add to AppKernel.php:

   new \Symfony\Bundle\MessengerBundle\MessengerBundle(),
   

2. Configure config.yml:

   messenger:
       transports:
           async: '%env(MESSENGER_TRANSPORT_DSN)%'
       routing:
           'App\Message\YourMessage': async
   

3. Create a Message Class:

   namespace App\Message;
   
   class YourMessage
   {
       public string $content;
       public function __construct(string $content) { $this->content = $content; }
   }
   

4. Dispatch a Message:

   use Symfony\Component\Messenger\MessageBusInterface;
   
   class YourController
   {
       public function __construct(private MessageBusInterface $bus) {}
   
       public function sendMessage()
       {
           $this->bus->dispatch(new YourMessage('Hello Messenger!'));
       }
   }
   

5. Run the Worker:

   php bin/console messenger:consume async -vv
   

First Use Case: Async Task Processing

Use this bundle to offload heavy or time-consuming tasks (e.g., sending emails, processing files) to a background worker. Example:

# config.yml
messenger:
    transports:
        async: 'doctrine://default'
    routing:
        'App\Message\ProcessFile': async

// Dispatch a file processing message
$bus->dispatch(new ProcessFile('/path/to/file.pdf'));

Implementation Patterns

Common Workflows

1. Message Dispatching

• Controller/Command: Inject MessageBusInterface and dispatch messages:

  $this->bus->dispatch(new YourMessage($data));
  

• Event Listeners: Dispatch messages in response to events:

  public function onKernelRequest(GetResponseEvent $event)
  {
      $this->bus->dispatch(new LogRequestEvent($event->getRequest()));
  }
  

2. Transport Configuration

• Doctrine Transport (for Symfony 2.7+):

  messenger:
      transports:
          doctrine: 'doctrine://default'
  

• Sync Transport (for testing):

  messenger:
      transports:
          sync: 'sync://'
  

• AMQP/RabbitMQ:

  messenger:
      transports:
          amqp: '%env(AMQP_DSN)%'
  

3. Middleware Integration

Add middleware to transform, validate, or log messages:

messenger:
    middleware:
        - 'App\Middleware\LogMessageMiddleware'

// src/Middleware/LogMessageMiddleware.php
public function handle($message, callable $next)
{
    \Log::info('Handling message: ' . get_class($message));
    return $next($message);
}

4. Retry Failed Messages

Configure retry logic in config.yml:

messenger:
    failure_transport: failed
    transports:
        failed: 'doctrine://default?queue_name=failed'

5. Command Bus for CLI Tasks

Use the CommandBus for CLI-specific messages:

messenger:
    buses:
        command.bus: [validation, 'App\Middleware\CliMiddleware']

Integration Tips

Laravel-Specific Adjustments

1. Service Container: Since Laravel uses its own DI container, manually bind the MessageBus:

   $this->app->bind(MessageBusInterface::class, function ($app) {
       return new MessageBus([
           new HandleMessageMiddleware(),
           // Add other middleware
       ]);
   });
   

2. Configuration Overrides: Override default config in config/messenger.php:

   'transports' => [
       'async' => [
           'dsn' => env('MESSENGER_TRANSPORT_DSN'),
           'options' => ['queue_name' => 'laravel_queue'],
       ],
   ],
   

3. Artisan Commands: Extend Symfony's MessengerCommand for Laravel:

   namespace App\Console\Commands;
   
   use Symfony\Component\Console\Command\Command;
   use Symfony\Component\Messenger\Command\ConsumeMessagesCommand;
   
   class ConsumeMessagesCommand extends ConsumeMessagesCommand
   {
       protected function getDefaultDefinition()
       {
           $inputDefinition = parent::getDefaultDefinition();
           $inputDefinition->addOption(new \Symfony\Component\Console\Input\InputOption(
               'time-limit',
               null,
               \InputOption::VALUE_OPTIONAL,
               'Time limit in seconds',
               30
           ));
           return $inputDefinition;
       }
   }
   

4. Queue Workers: Use Laravel's queue system as a transport:

   messenger:
       transports:
           laravel: 'doctrine://default?queue_name=laravel'
   

   Then, in a worker:

   $this->bus->dispatch(new YourMessage());
   

Gotchas and Tips

Pitfalls

1. Symfony 2.7+ Compatibility:

   • The bundle is a backport of Symfony 4.1's MessengerBundle. Some features (e.g., DIC autoconfiguration) may not work as expected. Manually bind services if needed.
   • Fix: Use explicit container bindings for MessageBus, Transport, and Router.

2. Doctrine Transport Quirks:

   • Doctrine transport may not work out-of-the-box with older Doctrine versions (<2.5). Ensure your doctrine/orm is compatible.
   • Fix: Upgrade Doctrine or use a different transport (e.g., sync:// for testing).

3. Message Serialization:

   • Messages must be serializable. Use __serialize() and __unserialize() for complex objects:

     public function __serialize() { return ['content' => $this->content]; }
     public function __unserialize(array $data) { $this->content = $data['content']; }
     

   • Fix: Implement Serializable interface or use Symfony's SerializerComponent.

4. Worker Process Management:

   • The messenger:consume command runs indefinitely. Use --time-limit to avoid hanging:

     php bin/console messenger:consume async --time-limit=60
     

   • Fix: Run workers in a process manager (e.g., Supervisor) with proper timeouts.

5. Configuration Overrides:

   • Laravel's config/messenger.php may not be auto-loaded. Ensure it’s included in config/app.php:

     'config' => [
         // ...
         'messenger' => __DIR__.'/messenger.php',
     ],
     

6. Middleware Order Matters:

   • Middleware runs in the order they’re defined. Place validation middleware early:

     messenger:
         middleware:
             - validation
             - 'App\Middleware\LogMiddleware'
     

Debugging Tips

1. Enable Debug Mode: Run the worker with -vv for verbose output:

   php bin/console messenger:consume async -vv
   

2. Check Failed Messages: Inspect the failed transport queue:

   php bin/console messenger:failed:list
   php bin/console messenger:failed:retry [message-id]
   

3. Log Middleware: Add a logging middleware to trace message flow:

   public function handle($message, callable $next)
   {
       \Log::debug('Message handled: ' . get_class($message));
       return $next($message);
   }
   

4. Test with Sync Transport: Use sync:// for local testing to avoid background workers:

   messenger:
       transports:
           sync: 'sync://'
       routing:
           'App\Message\*': sync
   

Extension Points

1. Custom Transports: Create a custom transport by implementing TransportInterface:

   class CustomTransport implements TransportInterface
   {
       public function get(): iterable { /* ... */ }
       public function ack(Envelope $envelope) { /* ... */ }
       public function reject(Envelope $envelope) { /* ... */ }
       public function send(Envelope $envelope) { /* ... */ }
   }
   

   Register it in config.yml:

   messenger:
       transports:
           custom: 'custom://'
   

2. Dynamic Routing: Use a RouterInterface to dynamically route messages:

   $router = new DynamicRouter([
       new CallbackRouter([
           YourMessage::class => function (MessageBusInterface $bus) {
               return $bus;
           },
       ]),
   ]);
   

3. Event Dispatching: Dispatch Symfony events from middleware:

   public function handle($message, callable $next)
   {
       $event = new MessageHandledEvent($message);
       $this