diff --git a/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php b/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php index 86bc28240..6bfa6fe15 100644 --- a/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php +++ b/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php @@ -15,6 +15,7 @@ use Ecotone\Messaging\MessageHeaders; use Ecotone\Messaging\Support\InvalidArgumentException; use Prooph\Common\Messaging\Message; +use Prooph\EventStore\EventStore; use Prooph\EventStore\Exception\ProjectionNotFound; use Prooph\EventStore\Exception\RuntimeException; use Prooph\EventStore\Pdo\Projection\MariaDbProjectionManager; @@ -36,8 +37,10 @@ */ class LazyProophProjectionManager implements ProjectionManager { - /** @var LazyProophProjectionManager[] */ + /** @var ProjectionManager[] */ private array $lazyInitializedProjectionManager = []; + /** @var EventStore[] */ + private array $initializedEventStores = []; /** * @param array $projectionSetupConfigurations @@ -53,19 +56,24 @@ public function __construct( private function getProjectionManager(): ProjectionManager { - $context = $this->lazyProophEventStore->getContextName(); - if (isset($this->lazyInitializedProjectionManager[$context])) { - return $this->lazyInitializedProjectionManager[$context]; + if ($this->eventSourcingConfiguration->isInMemory()) { + return $this->eventSourcingConfiguration->getInMemoryProjectionManager(); } + $context = $this->lazyProophEventStore->getContextName(); $eventStore = $this->getLazyProophEventStore(); + $innerEventStore = $eventStore->getEventStore(); + + if (isset($this->lazyInitializedProjectionManager[$context]) && $this->initializedEventStores[$context] === $innerEventStore) { + return $this->lazyInitializedProjectionManager[$context]; + } $this->lazyInitializedProjectionManager[$context] = match ($eventStore->getEventStoreType()) { - LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), - LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), - LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), - LazyProophEventStore::EVENT_STORE_TYPE_IN_MEMORY => $this->eventSourcingConfiguration->getInMemoryProjectionManager() + LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), + LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), + LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), }; + $this->initializedEventStores[$context] = $innerEventStore; return $this->lazyInitializedProjectionManager[$context]; } diff --git a/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php b/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php index c70391963..72cfcd081 100644 --- a/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php +++ b/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php @@ -64,6 +64,48 @@ public function test_building_synchronous_event_driven_projection(): void self::assertEquals([['ticket_id' => '124', 'ticket_type' => 'info']], $ecotone->sendQueryWithRouting('getInProgressTickets')); } + public function test_synchronous_projection_uses_the_current_transaction_after_reconnecting(): void + { + $connection = $this->getConnection(); + $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore( + containerOrAvailableServices: [new InProgressTicketList($connection), new TicketEventConverter(), DbalConnectionFactory::class => $this->getConnectionFactory()], + configuration: ServiceConfiguration::createWithDefaults() + ->withEnvironment('prod') + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::EVENT_SOURCING_PACKAGE])) + ->withNamespaces([ + 'Test\\Ecotone\\EventSourcing\\Fixture\\Ticket', + 'Test\\Ecotone\\EventSourcing\\Fixture\\TicketWithSynchronousEventDrivenProjection', + ]) + ->withExtensionObjects([ + EventSourcingConfiguration::createWithDefaults(), + ]), + pathToRootCatalog: __DIR__ . '/../../', + runForProductionEventStore: true + ); + + $ecotone->initializeProjection(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION); + $ecotone->sendCommand(new RegisterTicket('123', 'Johnny', 'alert')); + self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets')); + + $connection->close(); + $connection->beginTransaction(); + + try { + $ecotone->sendCommand(new CloseTicket('123')); + + self::assertTrue($connection->isTransactionActive()); + self::assertEquals([], $ecotone->sendQueryWithRouting('getInProgressTickets')); + } finally { + $connection->rollBack(); + } + + self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets')); + + $ecotone->sendCommand(new CloseTicket('123')); + + self::assertEquals([], $ecotone->sendQueryWithRouting('getInProgressTickets')); + } + public function test_synchronous_event_driven_projection_should_be_called_before_standard_event_handlers(): void { $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore(