add command bus service

DavidBadura · DavidBadura · commit 782a7aa81ae6 · 2025-02-16T13:31:58.000+01:00
diff --git a/docs/pages/configuration.md b/docs/pages/configuration.md
@@ -418,7 +418,7 @@ These handlers allow you to use your aggregates as command handlers.
 
 ```yaml
 patchlevel_event_sourcing:
-    aggregate_handlers: ~
+    command_bus: ~
 ```
 !!! note
 
@@ -428,6 +428,8 @@ Default the handlers will be registered for all buses.
 You can also specify a specific bus.
 Before you have to define the bus in the messenger configuration.
 
+### Symfony Messenger
+
 ```yaml
 framework:
     messenger:
@@ -443,9 +445,10 @@ And then you can specify the bus in the configuration.
 
 ```yaml
 patchlevel_event_sourcing:
-    aggregate_handlers:
-      bus: command.bus
+    command_bus:
+        service: command.bus
 ```
+
 ## Event Bus
 
 You can enable the event bus to listen for events and messages synchronously.
diff --git a/src/CommandBus/SymfonyCommandBus.php b/src/CommandBus/SymfonyCommandBus.php
@@ -0,0 +1,19 @@
+<?php
+
+namespace Patchlevel\EventSourcingBundle\CommandBus;
+
+use Patchlevel\EventSourcing\CommandBus\CommandBus;
+use Symfony\Component\Messenger\MessageBusInterface;
+
+final class SymfonyCommandBus implements CommandBus
+{
+    public function __construct(
+        private readonly MessageBusInterface $messageBus,
+    ) {
+    }
+
+    public function dispatch(object $command): void
+    {
+        $this->messageBus->dispatch($command);
+    }
+}
diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php
@@ -10,6 +10,7 @@
 /**
  * @psalm-type Config = array{
  *      event_bus: array{enabled: bool, type: string, service: string},
+ *      command_bus: array{enabled: bool, message_bus: string},
  *      subscription: array{
  *          store: array{type: string, service: string|null},
  *          retry_strategy: array{base_delay: int, delay_factor: int, max_attempts: int},
@@ -232,12 +233,26 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->end()
             ->end()
 
+            ->arrayNode('command_bus')
+                ->canBeEnabled()
+                ->addDefaultsIfNotSet()
+                ->children()
+                    ->scalarNode('service')->defaultNull()->end()
+                    ->booleanNode('register_aggregate_handlers')->defaultTrue()->end()
+                ->end()
+            ->end()
+
             ->arrayNode('aggregate_handlers')
                 ->canBeEnabled()
                 ->addDefaultsIfNotSet()
                 ->children()
                     ->scalarNode('bus')->defaultNull()->end()
                 ->end()
+                ->setDeprecated(
+                    'patchlevel/event-sourcing-bundle',
+                    '3.9',
+                    'The "%node%" option is deprecated and will be removed in 4.0. Use "patchlevel_event_sourcing.command_bus" instead.'
+                )
             ->end()
 
         ->end();
diff --git a/src/DependencyInjection/PatchlevelEventSourcingExtension.php b/src/DependencyInjection/PatchlevelEventSourcingExtension.php
@@ -21,6 +21,7 @@
 use Patchlevel\EventSourcing\Attribute\Subscriber;
 use Patchlevel\EventSourcing\Clock\FrozenClock;
 use Patchlevel\EventSourcing\Clock\SystemClock;
+use Patchlevel\EventSourcing\CommandBus\CommandBus;
 use Patchlevel\EventSourcing\Console\Command\DatabaseCreateCommand;
 use Patchlevel\EventSourcing\Console\Command\DatabaseDropCommand;
 use Patchlevel\EventSourcing\Console\Command\DebugCommand;
@@ -107,6 +108,7 @@
 use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberHelper;
 use Patchlevel\EventSourcingBundle\Attribute\AsListener;
 use Patchlevel\EventSourcingBundle\Command\StoreMigrateCommand;
+use Patchlevel\EventSourcingBundle\CommandBus\SymfonyCommandBus;
 use Patchlevel\EventSourcingBundle\DataCollector\EventSourcingCollector;
 use Patchlevel\EventSourcingBundle\DataCollector\MessageCollectorEventBus;
 use Patchlevel\EventSourcingBundle\Doctrine\DbalConnectionFactory;
@@ -135,7 +137,6 @@
 use Symfony\Component\DependencyInjection\Exception\InvalidArgumentException;
 use Symfony\Component\DependencyInjection\Extension\Extension;
 use Symfony\Component\DependencyInjection\Reference;
-
 use function class_exists;
 use function sprintf;
 
@@ -158,7 +159,7 @@ public function load(array $configs, ContainerBuilder $container): void
         $this->configureUpcaster($container);
         $this->configureSerializer($config, $container);
         $this->configureMessageDecorator($container);
-        $this->configureAggregateHandlers($config, $container);
+        $this->configureCommandBus($config, $container);
         $this->configureEventBus($config, $container);
         $this->configureConnection($config, $container);
         $this->configureStore($config, $container);
@@ -218,13 +219,36 @@ private function configureSerializer(array $config, ContainerBuilder $container)
     }
 
     /** @param Config $config */
-    private function configureAggregateHandlers(array $config, ContainerBuilder $container): void
+    private function configureCommandBus(array $config, ContainerBuilder $container): void
     {
+        if ($config['command_bus']['enabled'] && $config['aggregate_handlers']['enabled']) {
+            throw new InvalidArgumentException('Remove legacy aggregate_handlers configuration when using command_bus');
+        }
+
+        if ($config['command_bus']['enabled']) {
+            $container->register(SymfonyCommandBus::class)
+                ->setArguments([
+                    new Reference($config['command_bus']['service']),
+                ]);
+
+            $container->setAlias(CommandBus::class, SymfonyCommandBus::class);
+
+            $container->setParameter(
+                'patchlevel_event_sourcing.aggregate_handlers.bus',
+                $config['command_bus']['service']
+            );
+
+            return;
+        }
+
         if (!$config['aggregate_handlers']['enabled']) {
             return;
         }
 
-        $container->setParameter('patchlevel_event_sourcing.aggregate_handlers.bus', $config['aggregate_handlers']['bus']);
+        $container->setParameter(
+            'patchlevel_event_sourcing.aggregate_handlers.bus',
+            $config['aggregate_handlers']['bus']
+        );
     }
 
     /** @param Config $config */
diff --git a/tests/Unit/PatchlevelEventSourcingBundleTest.php b/tests/Unit/PatchlevelEventSourcingBundleTest.php
@@ -12,6 +12,7 @@
 use Patchlevel\EventSourcing\Aggregate\CustomId;
 use Patchlevel\EventSourcing\Clock\FrozenClock;
 use Patchlevel\EventSourcing\Clock\SystemClock;
+use Patchlevel\EventSourcing\CommandBus\CommandBus;
 use Patchlevel\EventSourcing\CommandBus\Handler\CreateAggregateHandler;
 use Patchlevel\EventSourcing\Console\Command\DatabaseCreateCommand;
 use Patchlevel\EventSourcing\Console\Command\DatabaseDropCommand;
@@ -70,6 +71,7 @@
 use Patchlevel\EventSourcing\Subscription\Store\SubscriptionStore;
 use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessorRepository;
 use Patchlevel\EventSourcingBundle\Command\StoreMigrateCommand;
+use Patchlevel\EventSourcingBundle\CommandBus\SymfonyCommandBus;
 use Patchlevel\EventSourcingBundle\DependencyInjection\PatchlevelEventSourcingExtension;
 use Patchlevel\EventSourcingBundle\EventBus\SymfonyEventBus;
 use Patchlevel\EventSourcingBundle\PatchlevelEventSourcingBundle;
@@ -547,6 +549,69 @@ public function testCommandHandler(): void
         self::assertEquals('command.bus', $tag['bus']);
     }
 
+    public function testCommandBusAndLegacyConfigurationNotAllowed(): void
+    {
+        $container = new ContainerBuilder();
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Remove legacy aggregate_handlers configuration when using command_bus');
+
+        $this->compileContainer(
+            $container,
+            [
+                'patchlevel_event_sourcing' => [
+                    'connection' => [
+                        'service' => 'doctrine.dbal.eventstore_connection',
+                    ],
+                    'aggregates' => [__DIR__ . '/../Fixtures'],
+                    'aggregate_handlers' => [
+                        'bus' => 'command.bus',
+                    ],
+                    'command_bus' => [
+                        'service' => 'command.bus',
+                    ],
+                ],
+            ]
+        );
+    }
+
+    public function testCommandBus(): void
+    {
+        $container = new ContainerBuilder();
+
+        $this->compileContainer(
+            $container,
+            [
+                'patchlevel_event_sourcing' => [
+                    'connection' => [
+                        'service' => 'doctrine.dbal.eventstore_connection',
+                    ],
+                    'aggregates' => [__DIR__ . '/../Fixtures'],
+                    'command_bus' => [
+                        'service' => 'command.bus',
+                    ],
+                ],
+            ]
+        );
+
+        $handler = $container->get('event_sourcing.handler.profile.create');
+
+        self::assertInstanceOf(CreateAggregateHandler::class, $handler);
+
+        $handler(new CreateProfile(CustomId::fromString('1')));
+
+        $definition = $container->getDefinition('event_sourcing.handler.profile.create');
+        $tags = $definition->getTag('messenger.message_handler');
+
+        self::assertCount(1, $tags);
+
+        $tag = $tags[0];
+
+        self::assertEquals(CreateProfile::class, $tag['handles']);
+        self::assertEquals('command.bus', $tag['bus']);
+        self::assertInstanceOf(SymfonyCommandBus::class, $container->get(CommandBus::class));
+    }
+
     public function testSnapshotStore(): void
     {
         $container = new ContainerBuilder();
@@ -1222,6 +1287,7 @@ private function compileContainer(ContainerBuilder $container, array $config): v
 
         $container->set('doctrine.dbal.eventstore_connection', $this->prophesize(Connection::class)->reveal());
         $container->set('event.bus', $this->prophesize(MessageBusInterface::class)->reveal());
+        $container->set('command.bus', $this->prophesize(MessageBusInterface::class)->reveal());
         $container->set('cache.default', $this->prophesize(CacheItemPoolInterface::class)->reveal());
         $container->set('event_dispatcher', $this->prophesize(EventDispatcherInterface::class)->reveal());
         $container->set('services_resetter', $this->prophesize(ServicesResetter::class)->reveal());
