Broadway Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require broadway/broadway-bundle
   

   Symfony Flex auto-configures the bundle with in-memory implementations (for development only).

2. First Use Case: Domain Event Handling Define a simple event:

   // src/Domain/Event/ExampleEvent.php
   namespace App\Domain\Event;
   
   use Broadway\Domain\DomainEvent;
   
   class ExampleEvent extends DomainEvent
   {
       public function __construct(private string $message)
       {
       }
   
       public function getMessage(): string
       {
           return $this->message;
       }
   }
   

3. Register Event Handlers

   // src/Domain/EventListener/ExampleEventListener.php
   namespace App\Domain\EventListener;
   
   use App\Domain\Event\ExampleEvent;
   use Broadway\EventHandling\EventListener;
   
   class ExampleEventListener implements EventListener
   {
       public function handle(object $event): void
       {
           if (!$event instanceof ExampleEvent) {
               return;
           }
           // Handle the event (e.g., log, update read model)
           info('Event handled: ' . $event->getMessage());
       }
   }
   

4. Configure in config/packages/broadway.yaml

   broadway:
       event_handlers:
           - App\Domain\EventListener\ExampleEventListener
   

5. Dispatch an Event

   use Broadway\Domain\DomainMessage;
   use Broadway\Domain\DomainRepository;
   use Broadway\EventHandling\EventBus;
   
   // In a service/controller
   $eventBus = $container->get(EventBus::class);
   $eventBus->dispatch(new ExampleEvent('Hello, Broadway!'));
   

Implementation Patterns

Domain-Driven Design (DDD) Workflow

1. Aggregate Root Management Define aggregates with invariants and domain events:

   // src/Domain/Aggregate/ExampleAggregate.php
   namespace App\Domain\Aggregate;
   
   use App\Domain\Event\ExampleEvent;
   use Broadway\Domain\AggregateRoot;
   
   class ExampleAggregate extends AggregateRoot
   {
       public function doSomething(string $input): void
       {
           $this->recordThat(new ExampleEvent($input));
       }
   }
   

2. Repository Integration Use DomainRepository to persist aggregates:

   // src/Infrastructure/Persistence/ExampleAggregateRepository.php
   namespace App\Infrastructure\Persistence;
   
   use App\Domain\Aggregate\ExampleAggregate;
   use Broadway\Domain\DomainRepository;
   
   class ExampleAggregateRepository extends DomainRepository
   {
       protected function getFQCN(): string
       {
           return ExampleAggregate::class;
       }
   }
   

3. Event Sourcing Pattern Replay events to reconstruct state:

   $repository = $container->get(ExampleAggregateRepository::class);
   $aggregate = $repository->getAggregateRoot($aggregateId);
   

4. Read Model Projections Use Projection to update read models:

   // src/Infrastructure/Projection/ExampleProjection.php
   namespace App\Infrastructure\Projection;
   
   use App\Domain\Event\ExampleEvent;
   use Broadway\ReadModel\Projection;
   
   class ExampleProjection implements Projection
   {
       public function __invoke(object $event): void
       {
           if ($event instanceof ExampleEvent) {
               // Update read model (e.g., Doctrine, Eloquent, or custom)
           }
       }
   }
   

5. Command Handling Dispatch commands to trigger domain logic:

   // src/Domain/Command/ExampleCommand.php
   namespace App\Domain\Command;
   
   use Broadway\CommandHandling\Command;
   
   class ExampleCommand implements Command
   {
       public function __construct(private string $input) {}
       public function getInput(): string { return $this->input; }
   }
   
   // Command handler
   use Broadway\CommandHandling\CommandHandler;
   
   class ExampleCommandHandler implements CommandHandler
   {
       public function handle(ExampleCommand $command): void
       {
           $aggregate = new ExampleAggregate();
           $aggregate->doSomething($command->getInput());
           $repository->save($aggregate);
       }
   }
   

Gotchas and Tips

Pitfalls

1. In-Memory Event Store

   • Default setup uses in-memory storage (lost on restart). For production, configure a persistent store (e.g., Doctrine, MongoDB):

     broadway:
         event_store:
             projectors: [App\Infrastructure\Projection\ExampleProjection]
             repository: broadway.event_store.doctrine
     

2. Event Listener Order

   • Listeners are executed in the order they are registered. Use priority annotations if order matters:

     use Broadway\EventHandling\EventListenerPriority;
     
     class HighPriorityListener implements EventListener, EventListenerPriority
     {
         public function getPriority(): int { return 10; }
         // ...
     }
     

3. Aggregate Identity

   • Ensure aggregate roots use a stable ID (e.g., UUID) to avoid replay issues during event sourcing.

4. Transaction Boundaries

   • Broadway does not manage transactions by default. Use Symfony’s transactional services or Doctrine transactions explicitly:

     $entityManager->beginTransaction();
     try {
         $repository->save($aggregate);
         $entityManager->commit();
     } catch (\Exception $e) {
         $entityManager->rollback();
         throw $e;
     }
     

5. Event Versioning

   • If events evolve, use EventVersion to handle backward compatibility:

     use Broadway\Domain\DomainEvent;
     
     class VersionedEvent extends DomainEvent
     {
         public function getVersion(): int { return 1; }
     }
     

Debugging Tips

1. Enable Debugging Configure logging in config/packages/monolog.yaml:

   monolog:
       handlers:
           broadway:
               type: stream
               path: "%kernel.logs_dir%/broadway.log"
               level: debug
   

2. Check Event Bus Verify events are dispatched and handled:

   $eventBus = $container->get(EventBus::class);
   $eventBus->dispatch(new ExampleEvent('Test'));
   // Check logs for confirmation
   

3. Projection Debugging Use ProjectionManager to inspect projections:

   $projectionManager = $container->get('broadway.read_model.projection_manager');
   $projectionManager->getProjection(ExampleProjection::class)->__invoke(new ExampleEvent('Test'));
   

Extension Points

1. Custom Event Store Implement EventStore interface for specialized storage (e.g., Kafka, RabbitMQ):

   use Broadway\EventStore\EventStore;
   
   class CustomEventStore implements EventStore
   {
       public function append(DomainMessage $domainMessage): void
       {
           // Custom logic
       }
       // ...
   }
   

2. Middleware for Commands/Events Add middleware to commands or events:

   use Broadway\CommandHandling\CommandMiddleware;
   use Broadway\EventHandling\EventMiddleware;
   
   class LoggingMiddleware implements CommandMiddleware, EventMiddleware
   {
       public function handle(object $message): void
       {
           info('Processing: ' . get_class($message));
       }
   }
   

3. Custom Metadata Attach metadata to events for auditing:

   use Broadway\Domain\Metadata;
   
   $metadata = new Metadata();
   $metadata->set('user_id', 123);
   $eventBus->dispatch(new ExampleEvent('Test', $metadata));
   

4. Symfony Messenger Bridge Integrate with Symfony Messenger for async processing:

   broadway:
       messenger:
           enabled: true
           transport: async