A Symfony bundle that wires Command, Query, and Event buses on top of Symfony Messenger. It auto-discovers handlers via PHP attributes, provides a configurable stamp pipeline, and ships with testing utilities and production-grade patterns.
Symfony Messenger is a powerful transport layer, but it leaves CQRS wiring as an exercise for the developer. This bundle fills the gap:
- Auto-discovery -- Annotate handlers with
#[AsCommandHandler],#[AsQueryHandler], or#[AsEventHandler]and they are registered automatically. No YAML tags, no manual wiring. - Stamp pipeline -- A composable
StampDeciderpipeline attaches retry policies, transport routing, serializer stamps, metadata, and dispatch-after-current-bus stamps per message type or per individual message class. - Type-safe buses -- Three dedicated buses (
CommandBus,QueryBus,EventBus) with distinct semantics: commands support sync/async dispatch, queries always return a result, events are fire-and-forget with zero-to-many handlers. - Testing utilities --
FakeCommandBus,FakeQueryBus,FakeEventBuswithassertDispatched(),assertNotDispatched(), and callback-based property assertions for fast, isolated unit tests.
flowchart LR
A[Your Code] --> B[CommandBus / QueryBus / EventBus]
B --> C[DispatchModeDecider]
C --> D[StampsDecider Pipeline]
D --> E1[RetryPolicy]
D --> E2[Transport]
D --> E3[Serializer]
D --> E4[Metadata]
D --> E5[DispatchAfterCurrentBus]
D --> F[Symfony Messenger]
F --> G[Handler]
| Capability | Raw Messenger | CQRS Bundle | Ecotone |
|---|---|---|---|
| Handler discovery | Manual YAML tags or #[AsMessageHandler] | #[AsCommandHandler] / #[AsQueryHandler] / #[AsEventHandler] with auto-discovery | Attribute-based with conventions |
| Type safety | Single MessageBusInterface | Separate CommandBus, QueryBus, EventBus with typed dispatch methods | Separate gateway interfaces |
| Bus abstraction | You build it | Three buses with sync/async routing, DispatchMode enum | Command/Query/Event buses built-in |
| Retry configuration | Per-transport YAML only | Per-message-class via RetryPolicy interface + resolver hierarchy | Per-endpoint via attributes |
| Testing support | InMemoryTransport | FakeBus implementations with assertDispatched() + callback assertions | Test support module |
| Async routing | routing YAML config | DispatchMode + #[Asynchronous] attribute + per-message transport mapping | Async via polled endpoints |
| Stamp pipeline | Manual stamp attachment | Composable StampDecider pipeline with priority ordering | Interceptors (before/after/around) |
| Event ordering | Not built-in | SequenceAware interface + AggregateSequenceStamp | Built-in aggregate versioning |
| Transactional outbox | Not built-in | OutboxStorage interface + DBAL implementation | Built-in with Doctrine |
| Sagas / Process managers | Not built-in | Not built-in | Built-in saga support |
| Event sourcing | Not built-in | Not built-in | Built-in event sourcing |
| OpenTelemetry | Not built-in | Bridge middleware with trace spans | Not built-in |
| Learning curve | Low (part of Symfony) | Low (thin layer over Messenger) | Moderate (own conventions) |
| Dependencies | Symfony only | Symfony Messenger | Ecotone framework |
Choose raw Messenger when your app has simple dispatch needs and you want zero additional dependencies. Choose this bundle when you want structured CQRS buses, per-message configuration, and testing utilities while staying close to Messenger. Choose Ecotone when you need sagas, event sourcing, or a full CQRS/ES framework.
Core
- CommandBus with sync/async dispatch and result extraction
- QueryBus with single-handler validation and typed results
- EventBus with zero-to-many handlers and fire-and-forget semantics
- Attribute-based handler discovery (
#[AsCommandHandler],#[AsQueryHandler],#[AsEventHandler]) - Handler interfaces optional -- attributes alone are sufficient
Stamp Pipeline
- Composable
StampDecidersystem with priority ordering (@api-- extend it yourself) - Per-message retry policies via
RetryPolicyinterface - Per-message transport routing with
TransportNamesStamporSendMessageToTransportsStamp - Per-message serializer stamps
- Per-message metadata stamps with correlation ID support
DispatchAfterCurrentBusStampcontrol per message
Patterns
- Causation ID propagation across nested dispatches
- Idempotency bridge (
IdempotencyStamptoDeduplicateStamp) - Event ordering with
SequenceAwareandAggregateSequenceStamp - Rate limiting via Symfony Rate Limiter integration
- Transactional outbox with DBAL storage and relay command
Developer Experience
FakeCommandBus,FakeQueryBus,FakeEventBusfor unit testingassertDispatched()/assertNotDispatched()with callback-based property assertionssomework:cqrs:generatescaffold command for messages and handlerssomework:cqrs:listhandler cataloguesomework:cqrs:debug-transportstransport diagnosticssomework:cqrs:health-checkfor monitoring
Observability
- OpenTelemetry bridge middleware (trace spans for dispatch and handling)
- PSR-3 structured logging across buses, deciders, and resolvers
Integration
- Symfony Flex recipe for zero-touch installation
CommandBusInterface,QueryBusInterface,EventBusInterfacefor DI and testing#[Asynchronous]attribute for transport routing without YAML config
- PHP 8.2 or newer.
- Symfony 7.2 or newer.
composer require somework/cqrs-bundleFlex automatically registers the bundle in config/bundles.php and creates a
commented config/packages/somework_cqrs.yaml with all available options.
Install the bundle via Composer:
composer require somework/cqrs-bundleThen register it manually in config/bundles.php:
return [
// ...SomeWork\CqrsBundle\SomeWorkCqrsBundle::class => ['all' => true],
];Create config/packages/somework_cqrs.yaml (see docs/flex-recipe/ for a
template with all available options).
Run the bundled console tooling to verify the bundle is registered:
bin/console somework:cqrs:listFlex Recipe: The recipe files are in
docs/flex-recipe/and are pending submission to symfony/recipes-contrib. Until published, manual bundle registration is required.
namespaceApp\Application\Command;
useSomeWork\CqrsBundle\Contract\Command;
finalclass CreateTask implements Command
{
publicfunction__construct(
publicreadonlystring$id,
publicreadonlystring$name,
) {}
}namespaceApp\Application\Command;
useSomeWork\CqrsBundle\Attribute\AsCommandHandler;
useSomeWork\CqrsBundle\Contract\CommandHandler;
#[AsCommandHandler(command: CreateTask::class)]
finalclass CreateTaskHandler implements CommandHandler
{
publicfunction__invoke(CreateTask$command): mixed
{
// Save task to database...returnnull;
}
}namespaceApp\Controller;
useApp\Application\Command\CreateTask;
useSomeWork\CqrsBundle\Contract\CommandBusInterface;
useSymfony\Component\HttpFoundation\JsonResponse;
useSymfony\Component\HttpFoundation\Request;
useSymfony\Component\Routing\Attribute\Route;
finalclass TaskController
{
#[Route('/tasks', methods: ['POST'])]
publicfunctioncreate(Request$request, CommandBusInterface$commandBus): JsonResponse
{
$data = $request->toArray();
$commandBus->dispatch(newCreateTask(
id: uuid_create(),
name: $data['name'],
));
returnnewJsonResponse(['status' => 'ok'], 201);
}
}Full documentation is available at somework.github.io/cqrs.
- Getting Started -- progressive tutorial from install to advanced patterns
- Usage Guide -- core patterns, dispatch modes, console commands
- Configuration Reference -- every
somework_cqrsoption explained - Testing Guide -- FakeBus, assertions, integration testing
- Production Guide -- deployment, workers, monitoring
- Troubleshooting -- common issues and solutions
- Upgrade Guide -- migration between versions
- Changelog
MIT. See LICENSE.