From 75b05e8fd8a3dfd471eae12741372b2835f5766e Mon Sep 17 00:00:00 2001 From: Alan Gabriel Bem Date: Mon, 23 Sep 2019 13:11:12 +0200 Subject: [PATCH] full list of changes below: - do not commit any changes after command has failed - try to cleanup transaction from (possibly) corrupted aggregates --- composer.json | 2 +- ...TransactionalPersistenceCommandHandler.php | 39 +++++++++++++- .../Domain/Event/Listener/CommandingTest.php | 4 +- ...sactionalPersistenceCommandHandlerTest.php | 52 ++++++++++++++++++- 4 files changed, 90 insertions(+), 7 deletions(-) diff --git a/composer.json b/composer.json index 7552c23e..1de3f4e6 100644 --- a/composer.json +++ b/composer.json @@ -29,7 +29,7 @@ "php": ">=7.1", "ext-PDO": "^7.1", "ext-igbinary": "^2.0|^3.0", - "ext-redis": "^4.0", + "ext-redis": "^4.0|^5.0", "doctrine/dbal": "^2.8", "psr/log": "~1.0", "ramsey/uuid": "^3.7" diff --git a/src/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandler.php b/src/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandler.php index 0dc8e677..5ada4cd7 100644 --- a/src/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandler.php +++ b/src/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandler.php @@ -14,6 +14,7 @@ namespace Streak\Infrastructure\CommandHandler; use Streak\Application; +use Streak\Domain\Event\Producer; use Streak\Infrastructure; /** @@ -39,8 +40,42 @@ public function __construct(Application\CommandHandler $handler, Infrastructure\ public function handle(Application\Command $command) : void { - $this->handler->handle($command); + $producersInTransactionBeforeCommand = $this->uow->uncommitted(); + try { + $this->handler->handle($command); + iterator_to_array($this->uow->commit()); + } catch (\Throwable $e) { + $producersInTransactionAfterCommand = $this->uow->uncommitted(); + $producersAddedToTransactionWithinCommand = $this->findProducersAddedWhileHandlingCommand($producersInTransactionBeforeCommand, $producersInTransactionAfterCommand); - iterator_to_array($this->uow->commit()); + foreach ($producersAddedToTransactionWithinCommand as $producer) { + $this->uow->remove($producer); + } + + throw $e; + } + } + + /** + * Basically finds producers that are present in $producersInTransactionBeforeCommand, but not in $producersInTransactionAfterCommand. + * + * @param Producer[] $producersInTransactionBeforeCommand + * @param Producer[] $producersInTransactionAfterCommand + * + * @return Producer[] + */ + private function findProducersAddedWhileHandlingCommand(array $producersInTransactionBeforeCommand, array $producersInTransactionAfterCommand) : array + { + $producersAddedToTransactionWithinCommand = []; + foreach ($producersInTransactionAfterCommand as $producerFromAfterCommandHandled) { + foreach ($producersInTransactionBeforeCommand as $producerFromBeforeCommandHandled) { + if ($producerFromAfterCommandHandled->producerId()->equals($producerFromBeforeCommandHandled->producerId())) { + continue 2; // next producer + } + } + $producersAddedToTransactionWithinCommand[] = $producerFromAfterCommandHandled; + } + + return $producersAddedToTransactionWithinCommand; } } diff --git a/tests/Domain/Event/Listener/CommandingTest.php b/tests/Domain/Event/Listener/CommandingTest.php index 58123f3f..a6b3e6ba 100644 --- a/tests/Domain/Event/Listener/CommandingTest.php +++ b/tests/Domain/Event/Listener/CommandingTest.php @@ -44,8 +44,8 @@ public function testReplayingEmptyStream() ->expects($this->exactly(2)) ->method('dispatch') ->withConsecutive( - new Command1(), - new Command3() + [new Command1()], + [new Command3()] ) ; $commander = new CommandingStub($this->bus); diff --git a/tests/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandlerTest.php b/tests/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandlerTest.php index f5e3f0d0..99c1a8a5 100644 --- a/tests/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandlerTest.php +++ b/tests/Infrastructure/CommandHandler/TransactionalPersistenceCommandHandlerTest.php @@ -87,8 +87,11 @@ public function setUp() $this->aggregateRoot1 = $this->getMockBuilder(Event\Sourced\AggregateRoot::class)->getMockForAbstractClass(); $this->aggregateRoot2 = $this->getMockBuilder(Event\Sourced\AggregateRoot::class)->getMockForAbstractClass(); - $this->aggregateRootId1 = $this->getMockBuilder(Domain\AggregateRoot\Id::class)->getMockForAbstractClass(); - $this->aggregateRootId2 = $this->getMockBuilder(Domain\AggregateRoot\Id::class)->getMockForAbstractClass(); + $this->aggregateRootId1 = new Infrastructure\CommandHandler\TransactionalPersistenceCommandHandlerTest\ProducerId('6448c6b7-bd5d-4a04-97a9-c1ad99008c04'); + $this->aggregateRootId2 = new Infrastructure\CommandHandler\TransactionalPersistenceCommandHandlerTest\ProducerId('9535ad9e-58c9-4bb6-82cf-843ae04f8f48'); + + $this->aggregateRoot1->expects($this->any())->method('producerId')->willReturn($this->aggregateRootId1); + $this->aggregateRoot2->expects($this->any())->method('producerId')->willReturn($this->aggregateRootId2); $this->producer1 = $this->getMockBuilder(Event\Producer::class)->getMockForAbstractClass(); $this->producer2 = $this->getMockBuilder(Event\Producer::class)->getMockForAbstractClass(); @@ -117,6 +120,42 @@ public function testHandlingCommand() $handler->handle($this->command); } + public function testException() + { + $exception = new \Exception(); + $this->expectExceptionObject($exception); + + $handler = new TransactionalPersistenceCommandHandler($this->handler, $this->uow); + + $this->handler + ->expects($this->once()) + ->method('handle') + ->with($this->command) + ->willThrowException($exception); + + $this->uow + ->expects($this->never()) + ->method('commit') + ; + + $this->uow + ->expects($this->exactly(2)) + ->method('uncommitted') + ->willReturnOnConsecutiveCalls( + [$this->aggregateRoot1], + [$this->aggregateRoot1, $this->aggregateRoot2] + ) + ; + + $this->uow + ->expects($this->once()) + ->method('remove') + ->with($this->aggregateRoot2) + ; + + $handler->handle($this->command); + } + public function committed(Event\Producer ...$producers) : \Generator { foreach ($producers as $producer) { @@ -124,3 +163,12 @@ public function committed(Event\Producer ...$producers) : \Generator } } } + +namespace Streak\Infrastructure\CommandHandler\TransactionalPersistenceCommandHandlerTest; + +use Streak\Domain\AggregateRoot; +use Streak\Domain\Id\UUID; + +class ProducerId extends UUID implements AggregateRoot\Id +{ +}