Broadway Laravel Package

Getting Started

Minimal Setup in Laravel

1. Install the Laravel Package:

   composer require nwidart/laravel-broadway
   

   Publish the config:

   php artisan vendor:publish --provider="Nwidart\Broadway\BroadwayServiceProvider"
   

2. Configure Event Store (config/broadway.php):

   'event_store' => [
       'connection' => 'mysql',
       'table' => 'broadway_events',
   ],
   

3. Run Migrations:

   php artisan migrate
   

4. First Use Case: Event-Sourced Aggregate Define an aggregate root (e.g., app/Domain/Invoice.php):

   use Broadway\Domain\DomainMessage;
   use Broadway\Domain\AggregateRoot;
   
   class Invoice extends AggregateRoot
   {
       public function createInvoice(string $invoiceId, float $amount)
       {
           $this->recordThat(new InvoiceCreated($invoiceId, $amount));
       }
   
       protected function apply(InvoiceCreated $event)
       {
           $this->invoiceId = $event->invoiceId;
           $this->amount = $event->amount;
       }
   }
   

   Register the aggregate in config/broadway.php:

   'aggregates' => [
       'App\\Domain\\Invoice' => [
           'repository' => 'App\\Domain\\InvoiceRepository',
       ],
   ],
   

Implementation Patterns

1. Command Handling Workflow

• Dispatch Commands:

  $commandBus = app('broadway.command_bus');
  $commandBus->dispatch(new CreateInvoiceCommand('INV-123', 100.0));
  

• Handle Commands (via SimpleCommandHandler):

  use Broadway\CommandHandling\SimpleCommandHandler;
  
  class CreateInvoiceHandler extends SimpleCommandHandler
  {
      public function handleCreateInvoiceCommand(CreateInvoiceCommand $command)
      {
          $invoice = new Invoice();
          $invoice->createInvoice($command->getInvoiceId(), $command->getAmount());
          $invoiceRepository = app('broadway.repository.invoice');
          $invoiceRepository->save($invoice);
      }
  }
  

• Register Handlers (via service provider):

  $this->app->bind('broadway.command_handler.create_invoice', function () {
      return new CreateInvoiceHandler();
  });
  

2. Event Sourcing with Projections

• Project Events to Read Models:

  use Broadway\ReadModel\Projector;
  
  class InvoiceProjector implements Projector
  {
      public function project(DomainMessage $domainMessage)
      {
          if (!$domainMessage instanceof InvoiceCreated) {
              return;
          }
          $readModel = new InvoiceReadModel(
              $domainMessage->invoiceId,
              $domainMessage->amount
          );
          $readModelRepository = app('broadway.read_model.invoice');
          $readModelRepository->save($readModel);
      }
  }
  

• Register Projector:

  $this->app->bind('broadway.projector.invoice', function () {
      return new InvoiceProjector();
  });
  

3. Testing with Broadway

• Scenario-Based Tests:

  use Broadway\CommandHandling\ScenarioTestCase;
  
  class InvoiceTest extends ScenarioTestCase
  {
      protected function provideScenarios()
      {
          return [
              'create_invoice' => [
                  new CreateInvoiceCommand('INV-123', 100.0),
                  [new InvoiceCreated('INV-123', 100.0)],
              ],
          ];
      }
  }
  

• Mock Event Store for Tests:

  $eventStore = new InMemoryEventStore();
  $this->setEventStore($eventStore);
  

4. Leveraging Processors for Async Work

• Process Events Asynchronously:

  use Broadway\Processor\EventProcessingResult;
  use Broadway\Processor\SimpleEventProcessor;
  
  class InvoiceEventProcessor extends SimpleEventProcessor
  {
      public function process(DomainMessage $domainMessage): EventProcessingResult
      {
          if ($domainMessage instanceof InvoiceCreated) {
              // Async logic (e.g., send email, update external API)
              return new EventProcessingResult();
          }
          return new EventProcessingResult();
      }
  }
  

• Register Processor:

  $this->app->bind('broadway.processor.invoice', function () {
      return new InvoiceEventProcessor();
  });
  

Gotchas and Tips

1. Event Store Quirks

• Playhead Management:

  • Always handle DuplicatePlayheadException when saving aggregates to avoid race conditions.
  • Use loadFromPlayhead() for optimistic locking:

    $invoice = $invoiceRepository->loadFromPlayhead($invoiceId, $playhead);
    

• Database Schema:

  • The broadway:event-store:schema:create command creates a table with playhead, aggregate_root_id, aggregate_type, recorded_on, and payload.
  • Ensure your payload column supports JSON serialization (e.g., LONGTEXT in MySQL).

2. Serialization Pitfalls

• Custom Serializers:

  • Broadway’s default serializer uses reflection. For complex objects, implement JsonSerializable or use the ReflectionSerializer:

    $serializer = new ReflectionSerializer();
    $serializer->serialize($event);
    

• Avoid Circular References:

  • Events containing circular references (e.g., EventA references EventB which references EventA) will fail serialization. Use DTOs or break cycles manually.

3. Command Handling Gotchas

• Handler Naming Convention:

  • Broadway uses handle<CommandName> (e.g., handleCreateInvoiceCommand). Forgetting the handle prefix will cause the handler to be ignored.

• Dependency Injection:

  • Handlers are instantiated via Laravel’s container. Use constructor injection for dependencies:

    public function __construct(private InvoiceRepository $invoiceRepository) {}
    

4. Testing Tips

• Reset Event Store Between Tests:

  • Clear the in-memory event store in setUp():

    protected function setUp(): void
    {
        parent::setUp();
        $this->eventStore = new InMemoryEventStore();
    }
    

• Verify Projections:

  • Use ReadModelTestCase to assert read model state:

    $this->assertReadModelEquals(
        new InvoiceReadModel('INV-123', 100.0),
        $this->readModelRepository->find('INV-123')
    );
    

5. Performance Considerations

• Batch Processing:

  • For high-throughput systems, use Broadway\Processor\BatchEventProcessor to process events in batches.

• Snapshot Aggregates:

  • Use the Snapshotting component to optimize loading large event streams:

    $repository = new SnapshottingAggregateRepository(
        $eventStore,
        new InMemorySnapshotStore(),
        $serializer
    );
    

6. Debugging

• Enable Auditing:

  • Log command success/failure by registering the CommandLogger:

    $this->app->bind('broadway.event_listener.command_logger', function () {
        return new CommandLogger(app('logger'));
    });
    

• Inspect Event Streams:

  • Use the EventStoreManagement interface to query events:

    $events = $eventStore->load($aggregateId);
    

7. Laravel-Specific Tips

• Service Provider Binding:

  • Bind repositories, projectors, and processors explicitly to avoid ambiguity:

    $this->app->bind('broadway.repository.invoice', function () {
        return new EventSourcingAggregateRepository(
            app('broadway.event_store'),
            Invoice::class,
            app('broadway.serializer')
        );
    });
    

• Artisan Commands:

  • Use Broadway’s CLI tools via Laravel’s Artisan:

    php artisan broadway:event-store:schema:create
    php artisan broadway:event-store:replay-projections
    

8. Extending Broadway

• Custom Event Store:

  • Implement Broadway\EventStore\EventStore for non-DBAL backends (e.g., Redis, MongoDB).

• Custom Serializer:

  • Extend Broadway\Serializer\Serializer for domain-specific needs:

    class CustomSerializer implements Serializer
    {
        public function serialize($object): string
        {
            return json_encode($object, JSON_UNESCAPED_SLASHES);
        }
    
        public function deserialize(string $serialized): object
        {
            return json_decode($serialized, true);
        }
    }