Messenger Laravel Package

Getting Started

Minimal Setup

1. Install the Package:

   composer require sulu/messenger
   

   Ensure your Laravel app uses Symfony Messenger (via spatie/laravel-messenger or direct symfony/messenger integration).

2. Register the Bundle (if not using Laravel’s service provider): Add to config/bundles.php (Symfony-style) or manually bind services in Laravel’s AppServiceProvider:

   // config/bundles.php (Symfony)
   Sulu\Messenger\Infrastructure\Symfony\HttpKernel\SuluMessengerBundle::class => ['all' => true],
   

3. First Use Case: Async Job with Locking Define a message and handler:

   // app/Messages/ProcessOrder.php
   class ProcessOrder implements MessageInterface {
       public function __construct(public string $orderId) {}
   }
   
   // app/Handlers/ProcessOrderHandler.php
   class ProcessOrderHandler {
       public function __invoke(ProcessOrder $message) {
           // Business logic here
           return new OrderProcessed($message->orderId);
       }
   }
   

   Dispatch with locking:

   use Sulu\Messenger\Infrastructure\Symfony\Messenger\LockMiddleware\LockStamp;
   use Symfony\Component\Messenger\Envelope;
   
   $this->bus->dispatch(
       new Envelope(new ProcessOrder('order-123'), [
           new LockStamp('order-lock-' . $message->orderId, 60.0) // 60s TTL
       ])
   );
   

4. Configure Transport (e.g., Doctrine or AMQP):

   # config/packages/messenger.yaml (Symfony-style)
   framework:
       messenger:
           transports:
               async: '%env(MESSENGER_TRANSPORT_DSN)%'
           routing:
               'App\Messages\ProcessOrder': async
   

Implementation Patterns

Core Workflows

1. Locking for Race Conditions

• Pattern: Use LockStamp for critical sections (e.g., inventory updates, payment processing).
• Example:

  $this->bus->dispatch(
      new Envelope(new UpdateInventory($productId), [
          new LockStamp('inventory-' . $productId, 30.0, true) // Auto-release
      ])
  );
  

• Laravel Integration: Wrap Symfony’s LockFactory in a Laravel service:

  $lockFactory = $this->container->get('lock.factory');
  $lock = $lockFactory->createLock('key', 60.0);
  

2. Doctrine Flush Control

• Pattern: Opt-in flushing via EnableFlushStamp to avoid global transactions.
• Example:

  $this->bus->dispatch(
      new Envelope(new SaveUserProfile($userId), [
          new EnableFlushStamp()
      ])
  );
  

• Laravel Note: Requires Doctrine ORM (not Eloquent). Use a custom adapter or bridge Doctrine to Laravel.

3. Exception Handling

• Pattern: Catch HandlerFailedException and map to HTTP responses (e.g., 422 for validation errors).
• Example:

  try {
      $this->bus->dispatch(new ValidateOrder($orderId));
  } catch (HandlerFailedException $e) {
      $originalException = $e->getOriginalException();
      return response()->json(['error' => $originalException->getMessage()], 422);
  }
  

4. Message Chaining (Sagas)

• Pattern: Chain messages with Envelope to model workflows (e.g., order → payment → shipping).
• Example:

  $envelope = new Envelope(new ProcessOrder($orderId));
  $envelope = $this->bus->dispatch($envelope);
  $this->bus->dispatch(new Envelope(new ShipOrder($orderId), $envelope->getStamps()));
  

Integration Tips

• Laravel Queues: Use Symfony Messenger with Laravel’s queue system by configuring a custom transport:

  // config/messenger.php
  'transports' => [
      'laravel_queue' => [
          'dsn' => 'laravel-queue://',
          'options' => [
              'queue' => 'default',
          ],
      ],
  ],
  

• Middleware Order: Add Sulu’s middlewares after Symfony’s HandleMessageMiddleware but before transports:

  # config/packages/messenger.yaml
  framework:
      messenger:
          middleware:
              - Sulu\Messenger\Infrastructure\Symfony\Messenger\UnpackExceptionMiddleware
              - Sulu\Messenger\Infrastructure\Symfony\Messenger\LockMiddleware\LockMiddleware
              - Sulu\Messenger\Infrastructure\Symfony\Messenger\FlushMiddleware\DoctrineFlushMiddleware
  

• Testing: Mock the bus and middlewares:

  $bus = $this->createMock(BusInterface::class);
  $bus->expects($this->once())
      ->method('dispatch')
      ->with($this->isInstanceOf(Envelope::class));
  

Gotchas and Tips

Pitfalls

1. Doctrine Dependency:

   • DoctrineFlushMiddleware requires Doctrine ORM (not Eloquent). Workaround:
     • Use a custom adapter or bridge Doctrine to Laravel.
     • Example: Extend DoctrineFlushMiddleware to support Eloquent:

       class EloquentFlushMiddleware implements MiddlewareInterface {
           public function handle(Envelope $envelope, HandleTrait $handle) {
               if ($envelope->last(EnableFlushStamp::class)) {
                   DB::commit(); // Laravel's DB facade
               }
               return $handle($envelope);
           }
       }
       

2. Lock Granularity:

   • Overly broad lock keys (e.g., global-lock) can block entire workflows. Use specific keys (e.g., user-123-profile).
   • Debugging: Check lock TTLs and auto-release settings to avoid deadlocks.

3. Exception Unpacking:

   • UnpackExceptionMiddleware exposes raw exceptions. Ensure your error handlers (e.g., API controllers) can process them:

     try {
         $this->bus->dispatch(new YourMessage());
     } catch (HandlerFailedException $e) {
         $error = $e->getOriginalException()->getMessage();
         // Log or return $error
     }
     

4. Middleware Order:

   • Incorrect order can break functionality. Test with:
     • UnpackExceptionMiddleware first to catch handler errors.
     • LockMiddleware before business logic.
     • DoctrineFlushMiddleware last to ensure flushing happens after all operations.

5. PHP 8.2+ Requirement:

   • Laravel 10+ supports PHP 8.2+, but older Laravel versions (e.g., 9.x) may conflict. Upgrade PHP if needed.

Debugging Tips

• Lock Issues:

  • Use Symfony’s LockFactory directly to inspect locks:

    $lockFactory = $this->container->get('lock.factory');
    $lock = $lockFactory->createLock('test-key');
    if (!$lock->acquire()) {
        // Handle lock failure
    }
    

  • Check for orphaned locks (e.g., crashes during message processing).

• Doctrine Flush:

  • Verify EnableFlushStamp is present in the Envelope:

    $envelope->last(EnableFlushStamp::class); // Returns stamp or null
    

  • Use Doctrine’s event listeners to log flush operations:

    // config/packages/doctrine.yaml
    doctrine:
        orm:
            event_listeners:
                flush_logger:
                    class: App\Doctrine\FlushLogger
                    tags: [doctrine.event_listener]
    

• Performance:

  • LockMiddleware: High contention can slow down processing. Monitor lock acquisition times.
  • DoctrineFlushMiddleware: Frequent flushing may impact performance. Batch operations where possible.

Extension Points

1. Custom Middleware:

   • Extend the bus with your own middleware (e.g., logging, metrics):

     class LogMiddleware implements MiddlewareInterface {
         public function handle(Envelope $envelope, HandleTrait $handle) {
             Log::info('Message dispatched', ['message' => $envelope->getMessage()]);
             return $handle($envelope);
         }
     }
     

   • Register in config/packages/messenger.yaml:

     framework:
         messenger:
             middleware:
                 - App\Middleware\LogMiddleware
     

2. Custom Stamps:

   • Add stamps for your use cases (e.g., RetryStamp):

     class RetryStamp implements StampInterface {
         public function __construct(public int $attempts) {}
     }