Merge pull request #92 from patchlevel/add-clock-decorator-and-upcast…

…er-configuration

add clock, decorator and upcaster configuration/di

composer.json

@@ -19,7 +19,7 @@
   ],
   "require": {
     "php": "~8.1.0",
-    "patchlevel/event-sourcing": "^2.0.0",
+    "patchlevel/event-sourcing": "^2.0.1",
     "symfony/config": "^4.4.34|^5.4.0|^6.0.0",
     "symfony/console": "^4.4.34|^5.4.0|^6.0.0",
     "symfony/dependency-injection": "^4.4.34|^5.4.0|^6.0.0",

src/DependencyInjection/Configuration.php

@@ -60,6 +60,13 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->scalarPrototype()->end()
             ->end()
 
+            ->arrayNode('clock')
+                ->addDefaultsIfNotSet()
+                ->children()
+                    ->scalarNode('freeze')->defaultNull()->end()
+                ->end()
+            ->end()
+
             ->arrayNode('watch_server')
                 ->addDefaultsIfNotSet()
                 ->children()
@@ -82,7 +89,6 @@ public function getConfigTreeBuilder(): TreeBuilder
                     ->children()
                         ->enumNode('type')->values(['psr6', 'psr16', 'custom'])->defaultValue('psr6')->end()
                         ->scalarNode('service')->end()
-                        ->scalarNode('batch_size')->defaultNull()->end()
                     ->end()
                 ->end()
             ->end()

src/DependencyInjection/PatchlevelEventSourcingExtension.php

@@ -4,6 +4,7 @@
 
 namespace Patchlevel\EventSourcingBundle\DependencyInjection;
 
+use DateTimeImmutable;
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\DriverManager;
 use Doctrine\Migrations\Configuration\Connection\ExistingConnection;
@@ -15,6 +16,9 @@
 use Doctrine\Migrations\Tools\Console\Command\ExecuteCommand;
 use Doctrine\Migrations\Tools\Console\Command\MigrateCommand;
 use Doctrine\Migrations\Tools\Console\Command\StatusCommand;
+use Patchlevel\EventSourcing\Clock\Clock;
+use Patchlevel\EventSourcing\Clock\FrozenClock;
+use Patchlevel\EventSourcing\Clock\SystemClock;
 use Patchlevel\EventSourcing\Console\Command\DatabaseCreateCommand;
 use Patchlevel\EventSourcing\Console\Command\DatabaseDropCommand;
 use Patchlevel\EventSourcing\Console\Command\DebugCommand;
@@ -27,6 +31,9 @@
 use Patchlevel\EventSourcing\Console\Command\ShowCommand;
 use Patchlevel\EventSourcing\Console\Command\WatchCommand;
 use Patchlevel\EventSourcing\Console\DoctrineHelper;
+use Patchlevel\EventSourcing\EventBus\Decorator\ChainMessageDecorator;
+use Patchlevel\EventSourcing\EventBus\Decorator\MessageDecorator;
+use Patchlevel\EventSourcing\EventBus\Decorator\RecordedOnDecorator;
 use Patchlevel\EventSourcing\EventBus\DefaultEventBus;
 use Patchlevel\EventSourcing\EventBus\EventBus;
 use Patchlevel\EventSourcing\EventBus\Listener;
@@ -52,6 +59,8 @@
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
 use Patchlevel\EventSourcing\Serializer\Hydrator\EventHydrator;
 use Patchlevel\EventSourcing\Serializer\Hydrator\MetadataEventHydrator;
+use Patchlevel\EventSourcing\Serializer\Upcast\Upcaster;
+use Patchlevel\EventSourcing\Serializer\Upcast\UpcasterChain;
 use Patchlevel\EventSourcing\Snapshot\Adapter\Psr16SnapshotAdapter;
 use Patchlevel\EventSourcing\Snapshot\Adapter\Psr6SnapshotAdapter;
 use Patchlevel\EventSourcing\Snapshot\DefaultSnapshotStore;
@@ -86,15 +95,17 @@ public function load(array $configs, ContainerBuilder $container): void
         $configuration = new Configuration();
 
         /**
-         * @var array{event_bus: ?array{type: string, service: string}, watch_server: array{enabled: bool, host: string}, connection: ?array{service: ?string, url: ?string}, store: array{schema_manager: string, type: string, options: array<string, mixed>}, aggregates: list<string>, events: list<string>, snapshot_stores: array<string, array{type: string, service: string, batch_size: ?int}>, migration: array{path: string, namespace: string}} $config
+         * @var array{event_bus: ?array{type: string, service: string}, watch_server: array{enabled: bool, host: string}, connection: ?array{service: ?string, url: ?string}, store: array{schema_manager: string, type: string, options: array<string, mixed>}, aggregates: list<string>, events: list<string>, snapshot_stores: array<string, array{type: string, service: string}>, migration: array{path: string, namespace: string}, clock: array{freeze: ?string}} $config
          */
         $config = $this->processConfiguration($configuration, $configs);
 
         if (!isset($config['connection'])) {
             return;
         }
 
+        $this->configureUpcaster($container);
         $this->configureSerializer($config, $container);
+        $this->configureMessageDecorator($container);
         $this->configureEventBus($config, $container);
         $this->configureProjection($container);
         $this->configureConnection($config, $container);
@@ -103,6 +114,7 @@ public function load(array $configs, ContainerBuilder $container): void
         $this->configureAggregates($config, $container);
         $this->configureCommands($container);
         $this->configureProfiler($container);
+        $this->configureClock($config, $container);
 
         if (class_exists(DependencyFactory::class)) {
             $this->configureMigration($config, $container);
@@ -141,6 +153,7 @@ private function configureSerializer(array $config, ContainerBuilder $container)
                 new Reference(EventRegistry::class),
                 new Reference(EventHydrator::class),
                 new Reference(Encoder::class),
+                new Reference(Upcaster::class),
             ]);
 
         $container->setAlias(EventSerializer::class, DefaultEventSerializer::class);
@@ -190,6 +203,32 @@ private function configureProjection(ContainerBuilder $container): void
         $container->setAlias(ProjectionHandler::class, MetadataAwareProjectionHandler::class);
     }
 
+    private function configureUpcaster(ContainerBuilder $container): void
+    {
+        $container->registerForAutoconfiguration(Upcaster::class)
+            ->addTag('event_sourcing.upcaster');
+
+        $container->register(UpcasterChain::class)
+            ->setArguments([new TaggedIteratorArgument('event_sourcing.upcaster')]);
+
+        $container->setAlias(Upcaster::class, UpcasterChain::class);
+    }
+
+    private function configureMessageDecorator(ContainerBuilder $container): void
+    {
+        $container->register(RecordedOnDecorator::class)
+            ->setArguments([new Reference(Clock::class)])
+            ->addTag('event_sourcing.message_decorator');
+
+        $container->registerForAutoconfiguration(MessageDecorator::class)
+            ->addTag('event_sourcing.message_decorator');
+
+        $container->register(ChainMessageDecorator::class)
+            ->setArguments([new TaggedIteratorArgument('event_sourcing.message_decorator')]);
+
+        $container->setAlias(MessageDecorator::class, ChainMessageDecorator::class);
+    }
+
     /**
      * @param array{connection: array{url: ?string, service: ?string}} $config
      */
@@ -256,7 +295,7 @@ private function configureStorage(array $config, ContainerBuilder $container): v
     }
 
     /**
-     * @param array{snapshot_stores: array<string, array{type: string, service: string, batch_size: ?int}>} $config
+     * @param array{snapshot_stores: array<string, array{type: string, service: string}>} $config
      */
     private function configureSnapshots(array $config, ContainerBuilder $container): void
     {
@@ -306,6 +345,7 @@ private function configureAggregates(array $config, ContainerBuilder $container)
                 new Reference(Store::class),
                 new Reference(EventBus::class),
                 new Reference(SnapshotStore::class),
+                new Reference(MessageDecorator::class),
             ]);
 
         $container->setAlias(RepositoryManager::class, DefaultRepositoryManager::class);
@@ -481,4 +521,22 @@ private function configureProfiler(ContainerBuilder $container): void
             ])
             ->addTag('data_collector', ['template' => '@PatchlevelEventSourcing/Collector/template.html.twig']);
     }
+
+    /**
+     * @param array{clock: array{freeze: ?string}} $config
+     */
+    private function configureClock(array $config, ContainerBuilder $container): void
+    {
+        if ($config['clock']['freeze'] === null) {
+            $container->register(SystemClock::class);
+            $container->setAlias(Clock::class, SystemClock::class);
+
+            return;
+        }
+
+        $container->register(FrozenClock::class)
+            ->setArguments([new DateTimeImmutable($config['clock']['freeze'])]);
+
+        $container->setAlias(Clock::class, FrozenClock::class);
+    }
 }

tests/Unit/PatchlevelEventSourcingBundleTest.php

@@ -8,6 +8,9 @@
 use Doctrine\Migrations\Tools\Console\Command\ExecuteCommand;
 use Doctrine\Migrations\Tools\Console\Command\MigrateCommand;
 use Doctrine\Migrations\Tools\Console\Command\StatusCommand;
+use Patchlevel\EventSourcing\Clock\Clock;
+use Patchlevel\EventSourcing\Clock\FrozenClock;
+use Patchlevel\EventSourcing\Clock\SystemClock;
 use Patchlevel\EventSourcing\Console\Command\DatabaseCreateCommand;
 use Patchlevel\EventSourcing\Console\Command\DatabaseDropCommand;
 use Patchlevel\EventSourcing\Console\Command\DebugCommand;
@@ -24,12 +27,10 @@
 use Patchlevel\EventSourcing\EventBus\SymfonyEventBus;
 use Patchlevel\EventSourcing\Metadata\AggregateRoot\AggregateRootRegistry;
 use Patchlevel\EventSourcing\Metadata\Event\EventRegistry;
-use Patchlevel\EventSourcing\Projection\DefaultProjectionHandler;
 use Patchlevel\EventSourcing\Projection\MetadataAwareProjectionHandler;
 use Patchlevel\EventSourcing\Projection\ProjectionHandler;
 use Patchlevel\EventSourcing\Repository\DefaultRepository;
 use Patchlevel\EventSourcing\Repository\DefaultRepositoryManager;
-use Patchlevel\EventSourcing\Repository\Repository;
 use Patchlevel\EventSourcing\Repository\RepositoryManager;
 use Patchlevel\EventSourcing\Schema\SchemaManager;
 use Patchlevel\EventSourcing\Snapshot\Adapter\Psr16SnapshotAdapter;
@@ -98,6 +99,7 @@ public function testMinimalConfig()
         self::assertInstanceOf(AggregateRootRegistry::class, $container->get(AggregateRootRegistry::class));
         self::assertInstanceOf(DefaultRepositoryManager::class, $container->get(RepositoryManager::class));
         self::assertInstanceOf(EventRegistry::class, $container->get(EventRegistry::class));
+        self::assertInstanceOf(SystemClock::class, $container->get(Clock::class));
     }
 
     public function testConnectionService()
@@ -566,6 +568,30 @@ public function testMigrations()
         self::assertInstanceOf(StatusCommand::class, $container->get('event_sourcing.command.migration_status'));
     }
 
+    public function testFrozenClock()
+    {
+        $container = new ContainerBuilder();
+
+        $this->compileContainer(
+            $container,
+            [
+                'patchlevel_event_sourcing' => [
+                    'connection' => [
+                        'service' => 'doctrine.dbal.eventstore_connection',
+                    ],
+                    'clock' => [
+                        'freeze' => '2020-01-01 22:00:00',
+                    ],
+                ],
+            ]
+        );
+
+        $clock = $container->get(Clock::class);
+
+        self::assertInstanceOf(FrozenClock::class, $clock);
+        self::assertSame('2020-01-01 22:00:00', $clock->now()->format('Y-m-d H:i:s'));
+    }
+
     public function testFullBuild()
     {
         $container = new ContainerBuilder();