From 2c35021d4f73712cb902618d5fe9a600a9e3cf74 Mon Sep 17 00:00:00 2001 From: David Badura Date: Mon, 5 Oct 2026 12:10:59 +0200 Subject: [PATCH 1/3] Pass argument resolvers to the subscription engine Since event-sourcing 4.0 the argument resolvers are passed to the DefaultSubscriptionEngine instead of the MetadataSubscriberAccessorRepository. The bundle still passed them to the repository, where they were silently ignored, so the LookupResolver and custom resolvers were never used. Also add the missing EventRegistry argument when migrating to a TaggableDoctrineDbalStore. --- .../PatchlevelEventSourcingExtension.php | 3 +- .../PatchlevelEventSourcingBundleTest.php | 38 +++++++++++++++---- 2 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/DependencyInjection/PatchlevelEventSourcingExtension.php b/src/DependencyInjection/PatchlevelEventSourcingExtension.php index e7e4b488..cd2add8b 100644 --- a/src/DependencyInjection/PatchlevelEventSourcingExtension.php +++ b/src/DependencyInjection/PatchlevelEventSourcingExtension.php @@ -491,7 +491,6 @@ static function (ChildDefinition $definition): void { ->setArguments([ new TaggedIteratorArgument('event_sourcing.subscriber'), new Reference(SubscriberMetadataFactory::class), - new TaggedIteratorArgument('event_sourcing.argument_resolver'), ]); $container->setAlias(SubscriberAccessorRepository::class, MetadataSubscriberAccessorRepository::class); @@ -515,6 +514,7 @@ static function (ChildDefinition $definition): void { new Reference('logger', ContainerInterface::NULL_ON_INVALID_REFERENCE), new Reference(Cleaner::class), ]) + ->setArgument('$argumentResolvers', new TaggedIteratorArgument('event_sourcing.argument_resolver')) ->addTag('monolog.logger', ['channel' => 'event_sourcing']); $container->setAlias(SubscriptionEngine::class, DefaultSubscriptionEngine::class); @@ -849,6 +849,7 @@ private function configureStoreMigration(array $config, ContainerBuilder $contai ->setArguments([ new Reference('event_sourcing.dbal_connection'), new Reference(EventSerializer::class), + new Reference(EventRegistry::class), new Reference(HeadersSerializer::class), new Reference('event_sourcing.clock'), $config['store']['migrate_to_new_store']['options'], diff --git a/tests/Unit/PatchlevelEventSourcingBundleTest.php b/tests/Unit/PatchlevelEventSourcingBundleTest.php index 3cf46a06..d0af33e3 100644 --- a/tests/Unit/PatchlevelEventSourcingBundleTest.php +++ b/tests/Unit/PatchlevelEventSourcingBundleTest.php @@ -76,6 +76,7 @@ use Patchlevel\EventSourcing\Store\ReadOnlyStore; use Patchlevel\EventSourcing\Store\Store; use Patchlevel\EventSourcing\Store\StreamDoctrineDbalStore; +use Patchlevel\EventSourcing\Store\TaggableDoctrineDbalStore; use Patchlevel\EventSourcing\Subscription\Cleanup\Cleaner; use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DbalCleanupTaskHandler; use Patchlevel\EventSourcing\Subscription\Cleanup\DefaultCleaner; @@ -94,7 +95,6 @@ use Patchlevel\EventSourcing\Subscription\Store\DoctrineSubscriptionStore; use Patchlevel\EventSourcing\Subscription\Store\InMemorySubscriptionStore; use Patchlevel\EventSourcing\Subscription\Store\SubscriptionStore; -use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessorRepository; use Patchlevel\EventSourcingBundle\DependencyInjection\PatchlevelEventSourcingExtension; use Patchlevel\EventSourcingBundle\EventBus\SymfonyEventBus; use Patchlevel\EventSourcingBundle\Normalizer\SymfonyExtension; @@ -138,6 +138,8 @@ use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter; use Symfony\Component\Messenger\MessageBusInterface; +use function array_filter; + final class PatchlevelEventSourcingBundleTest extends TestCase { public function testEmptyConfig(): void @@ -354,6 +356,26 @@ public function testMigrateStore(): void ); } + public function testMigrateToTaggableStore(): void + { + $container = new ContainerBuilder(); + + $this->compileContainer( + $container, + [ + 'patchlevel_event_sourcing' => [ + 'connection' => ['service' => 'doctrine.dbal.eventstore_connection'], + 'store' => [ + 'migrate_to_new_store' => ['type' => 'dbal_taggable'], + ], + ], + ], + ); + + self::assertInstanceOf(StreamDoctrineDbalStore::class, $container->get(Store::class)); + self::assertInstanceOf(TaggableDoctrineDbalStore::class, $container->get('event_sourcing.store.new_store')); + } + public function testSymfonyEventBus(): void { $eventBus = $this->createMock(MessageBusInterface::class); @@ -1380,14 +1402,14 @@ public function testAutoconfigureArgumentResolver(): void ); self::assertTrue($container->getDefinition(DummyArgumentResolver::class)->hasTag('event_sourcing.argument_resolver')); - self::assertInstanceOf( - TaggedIteratorArgument::class, - $container->getDefinition(MetadataSubscriberAccessorRepository::class)->getArgument(2), - ); - self::assertEquals( - 'event_sourcing.argument_resolver', - $container->getDefinition(MetadataSubscriberAccessorRepository::class)->getArgument(2)->getTag(), + + $argumentResolvers = array_filter( + $container->getDefinition(DefaultSubscriptionEngine::class)->getArguments(), + static fn (mixed $argument): bool => $argument instanceof TaggedIteratorArgument + && $argument->getTag() === 'event_sourcing.argument_resolver', ); + + self::assertCount(1, $argumentResolvers); } public function testRetryStrategy(): void From 6dea4078ffb909c384fe42ebb5bdc10a875a33f4 Mon Sep 17 00:00:00 2001 From: David Badura Date: Mon, 5 Oct 2026 12:18:39 +0200 Subject: [PATCH 2/3] Add a dedicated event dispatcher for the subscription engine The engine dispatches events like OnHandleMessageError on its own dispatcher and registers internal listeners on it, so we provide a separate service instead of the global event dispatcher. Users can hook in via AsEventListener with the dispatcher option. This also lets us pass the argument resolvers positionally. --- docs/configuration.md | 23 +++++++++++++ .../PatchlevelEventSourcingExtension.php | 6 +++- .../PatchlevelEventSourcingBundleTest.php | 33 +++++++++++++++---- 3 files changed, 54 insertions(+), 8 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 43b9d18b..7e52ae5b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -456,6 +456,29 @@ patchlevel_event_sourcing: gap_detection: detection_window: 'PT5M' ``` +### Engine Events + +The subscription engine dispatches events while processing, e.g. `OnHandleMessageError` or `OnResult`. +It uses its own event dispatcher `event_sourcing.subscription.event_dispatcher`, +so you have to pass it to the `AsEventListener` attribute. + +```php +use Patchlevel\EventSourcing\Subscription\Engine\Event\OnHandleMessageError; +use Symfony\Component\EventDispatcher\Attribute\AsEventListener; + +#[AsEventListener(dispatcher: 'event_sourcing.subscription.event_dispatcher')] +final class SubscriptionErrorListener +{ + public function __invoke(OnHandleMessageError $event): void + { + // logging, metrics, ... + } +} +``` +:::note +You can find all available events in the [library documentation](https://event-sourcing.patchlevel.io/latest/subscription/). +::: + ## Command Bus You can enable the command bus integration to use your aggregates as command handlers. diff --git a/src/DependencyInjection/PatchlevelEventSourcingExtension.php b/src/DependencyInjection/PatchlevelEventSourcingExtension.php index cd2add8b..354df807 100644 --- a/src/DependencyInjection/PatchlevelEventSourcingExtension.php +++ b/src/DependencyInjection/PatchlevelEventSourcingExtension.php @@ -153,6 +153,7 @@ use Symfony\Component\DependencyInjection\Extension\Extension; use Symfony\Component\DependencyInjection\Parameter; use Symfony\Component\DependencyInjection\Reference; +use Symfony\Component\EventDispatcher\EventDispatcher; use function class_exists; use function sprintf; @@ -505,6 +506,8 @@ static function (ChildDefinition $definition): void { $container->setAlias(Cleaner::class, DefaultCleaner::class); + $container->register('event_sourcing.subscription.event_dispatcher', EventDispatcher::class); + $container->register(DefaultSubscriptionEngine::class) ->setArguments([ new Reference(MessageLoader::class), @@ -513,8 +516,9 @@ static function (ChildDefinition $definition): void { new Reference(RetryStrategyRepository::class), new Reference('logger', ContainerInterface::NULL_ON_INVALID_REFERENCE), new Reference(Cleaner::class), + new Reference('event_sourcing.subscription.event_dispatcher'), + new TaggedIteratorArgument('event_sourcing.argument_resolver'), ]) - ->setArgument('$argumentResolvers', new TaggedIteratorArgument('event_sourcing.argument_resolver')) ->addTag('monolog.logger', ['channel' => 'event_sourcing']); $container->setAlias(SubscriptionEngine::class, DefaultSubscriptionEngine::class); diff --git a/tests/Unit/PatchlevelEventSourcingBundleTest.php b/tests/Unit/PatchlevelEventSourcingBundleTest.php index d0af33e3..6d46b1ae 100644 --- a/tests/Unit/PatchlevelEventSourcingBundleTest.php +++ b/tests/Unit/PatchlevelEventSourcingBundleTest.php @@ -135,11 +135,10 @@ use Symfony\Component\DependencyInjection\Definition; use Symfony\Component\DependencyInjection\Dumper\XmlDumper; use Symfony\Component\DependencyInjection\Reference; +use Symfony\Component\EventDispatcher\EventDispatcher; use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter; use Symfony\Component\Messenger\MessageBusInterface; -use function array_filter; - final class PatchlevelEventSourcingBundleTest extends TestCase { public function testEmptyConfig(): void @@ -1403,13 +1402,33 @@ public function testAutoconfigureArgumentResolver(): void self::assertTrue($container->getDefinition(DummyArgumentResolver::class)->hasTag('event_sourcing.argument_resolver')); - $argumentResolvers = array_filter( - $container->getDefinition(DefaultSubscriptionEngine::class)->getArguments(), - static fn (mixed $argument): bool => $argument instanceof TaggedIteratorArgument - && $argument->getTag() === 'event_sourcing.argument_resolver', + $argument = $container->getDefinition(DefaultSubscriptionEngine::class)->getArgument(7); + + self::assertInstanceOf(TaggedIteratorArgument::class, $argument); + self::assertEquals('event_sourcing.argument_resolver', $argument->getTag()); + } + + public function testSubscriptionEventDispatcher(): void + { + $container = new ContainerBuilder(); + + $this->compileContainer( + $container, + [ + 'patchlevel_event_sourcing' => [ + 'connection' => ['service' => 'doctrine.dbal.eventstore_connection'], + ], + ], ); - self::assertCount(1, $argumentResolvers); + self::assertEquals( + new Reference('event_sourcing.subscription.event_dispatcher'), + $container->getDefinition(DefaultSubscriptionEngine::class)->getArgument(6), + ); + self::assertInstanceOf( + EventDispatcher::class, + $container->get('event_sourcing.subscription.event_dispatcher'), + ); } public function testRetryStrategy(): void From ea59496910aa601483e64dc021435a34c6f19d02 Mon Sep 17 00:00:00 2001 From: David Badura Date: Mon, 5 Oct 2026 12:32:34 +0200 Subject: [PATCH 3/3] Use relative link to the library docs --- docs/configuration.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 7e52ae5b..e9c136be 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -284,7 +284,7 @@ patchlevel_event_sourcing: :::tip You can find out more about subscriptions in the library -[documentation](https://event-sourcing.patchlevel.io/latest/subscription/). +[documentation](/docs/event-sourcing/latest/subscription). ::: ### Store @@ -476,7 +476,7 @@ final class SubscriptionErrorListener } ``` :::note -You can find all available events in the [library documentation](https://event-sourcing.patchlevel.io/latest/subscription/). +You can find all available events in the [library documentation](/docs/event-sourcing/latest/subscription). ::: ## Command Bus