Merge pull request #17 from codeliner/patch-12

Patch 12

README.md

@@ -5,7 +5,7 @@ CQRS + ES for ZF2
 [![Build Status](https://travis-ci.org/prooph/proophessor.svg?branch=master)](https://travis-ci.org/prooph/proophessor)
 [![Coverage Status](https://coveralls.io/repos/prooph/proophessor/badge.svg?branch=master)](https://coveralls.io/r/prooph/proophessor?branch=master)
 
-Proophessor combines [prooph/service-bus](https://github.com/prooph/service-bus), [proop/event-store](https://github.com/prooph/event-store) and [prooph/event-sourcing](https://github.com/prooph/event-sourcing) in a single ZF2 module. Goal is to simplify the set up process for a full featured CQRS + ES system.
+Proophessor combines [prooph/service-bus](https://github.com/prooph/service-bus), [proop/event-store](https://github.com/prooph/event-store) and [prooph/event-sourcing](https://github.com/prooph/event-sourcing) in a single ZF2 module to simplify the set up process for a full featured CQRS + ES system.
 
 ## Key Facts
 - [x] CQRS messaging tools
@@ -313,20 +313,10 @@ a secure way to convert a command to a remote message which can be send to a rem
 Proophessor doesn't work with serializers or annotations to help you with type mapping, because they slow down the system
 and add complexity.
 
-### Transaction Handling
-
-Proophessor automatically handles transactions for you. Each time you dispatch a command a new transaction is started.
-A successful dispatch commits the transaction and an error causes a rollback. `Proophessor only opens one transaction.`
-If you work with a process manager which listens on synchronous dispatched events and the process manager dispatches
-follow up commands, these commands are handled `in the same transaction as the first command`. If a follow up command fails
-the transaction is completely rolled back including `all recorded events` and potential `changes in the read model`.
-Again, this only happens if your events are dispatched `synchronous` and if the event store and the read model `share the same
-database connection`.
-
 ### Command Flow
 
 A command is normally dispatched from inside a MVC controller action or a process manager (when an domain event causes a follow up command).
-Because we don't need to care about transaction handling dispatching a command is relative simple:
+Dispatching a command is relatively simple:
 
 ```php
 <?php
@@ -757,6 +747,50 @@ final class UserProjector
 }
 ```
 
+### Transaction Handling
+
+Proophessor automatically handles transactions for you. Each time you dispatch a command a new transaction is started.
+A successful dispatch commits the transaction and an error causes a rollback. `Proophessor only opens one transaction.`
+If you work with a process manager which listens on synchronous dispatched events and the process manager dispatches
+follow up commands, these commands are handled `in the same transaction as the first command`. If a follow up command fails
+the transaction is completely rolled back including `all recorded events` and potential `changes in the read model`.
+Again, this only happens if your events are dispatched `synchronous` and if the event store and the read model `share the same
+database connection`.
+
+The TransactionManager is responsible for handling these scenarios. It makes use of the action event systems provided by
+prooph/service-bus and prooph/event-store to seamlessly integrate transaction handling.
+
+#### Begin, Commit, Rollback
+The TransactionManager registers a listener on the `Prooph\ServiceBus\Process\CommandDispatch::INITIALIZE` action event with a low
+priority of `-1000` to begin a new transaction if no one is already started and only if the command extends `Prooph\Common\Messaging\Command` and
+does not implement `Prooph\Proophessor\EventStore\AutoCommitCommand`. The latter is a marker interface to tell the TransactionManager that it
+should ignore such a command and all domain events caused by it.
+
+Furthermore, the TransactionManager registers a listener on the `Prooph\ServiceBus\Process\CommandDispatch::FINALIZE` action event with a high
+priority of `1000`. In the listener the TransactionManager decides either if it commits the current transaction or if it performs a rollback
+depending on the current CommandDispatch.
+
+A rollback is performed if `CommandDispatch::getException` returns a caught exception. With that in mind you
+can influence the rollback behaviour by registering an own listener on `Prooph\ServiceBus\Process\CommandDispatch::ERROR`, handle a caught exception of
+your own (f.e. retry current command, translate exception into DomainEvent, etc.) and unset the exception in the CommandDispatch by invoking
+`CommandDispatch::setException` with `NULL` as argument.
+
+If the TransactionManager has an active transaction and no exception was caught during dispatch (or was unset) the transaction gets committed.
+
+#### Event Dispatch and Status
+The TransactionManager has a second job which is related to handling the transaction. It adds meta information to each recorded domain event
+namely the UUID of the current command - referenced as `causation_id`, the message name of the current command - referenced as `causation_name` and
+a `dispatch_status`. The latter can either be `0 = event dispatch not started` or `2 = event dispatch was successful`.
+
+To determine the dispatch status of a domain event the TransactionManager checks if the recorded event should be dispatched synchronous or
+asynchronous based on the marker interface: `Prooph\Proophessor\EventStore\IsProcessedAsync`.
+
+All synchronous domain events are forwarded to the EventBus by the TransactionManager and the dispatch status is set to `2`
+(a failing event dispatch causes a transaction rollback, so that status will never be set for sync events).
+
+For all asynchronous domain events the dispatch status is set to `0` and they are not forwarded to the EventBus!
+
+
 Support
 -------
 

spec/EventStore/TransactionManagerSpec.php

@@ -39,8 +39,7 @@ function it_is_initializable(ActionEventDispatcher $actionEventDispatcher)
     function it_acts_as_a_message_bus_plugin_to_monitor_command_dispatch(ActionEventDispatcher $actionEventDispatcher, ListenerHandler $listenerHandler)
     {
         $actionEventDispatcher->attachListener(CommandDispatch::INITIALIZE, [$this, 'onInitialize'], -1000)->willReturn($listenerHandler);
-        $actionEventDispatcher->attachListener(CommandDispatch::HANDLE_ERROR, [$this, 'onError'])->willReturn($listenerHandler);
-        $actionEventDispatcher->attachListener(CommandDispatch::FINALIZE, [$this, 'onFinalize'])->willReturn($listenerHandler);
+        $actionEventDispatcher->attachListener(CommandDispatch::FINALIZE, [$this, 'onFinalize'], 1000)->willReturn($listenerHandler);
 
         $this->attach($actionEventDispatcher);
     }
@@ -69,7 +68,9 @@ function it_performs_a_rollback_on_command_dispatch_error(EventStore $eventStore
 
         $eventStore->rollback()->shouldBeCalled();
 
-        $this->onError($commandDispatch);
+        $commandDispatch->getException()->willReturn(new \Exception());
+
+        $this->onFinalize($commandDispatch);
     }
 
     function it_does_not_perform_a_rollback_when_command_implements_auto_commit_command(EventStore $eventStore, CommandDispatch $commandDispatch, AutoCommitCommandStub $autoCommitCommand)
@@ -82,14 +83,18 @@ function it_does_not_perform_a_rollback_when_command_implements_auto_commit_comm
 
         $commandDispatch->getCommand()->willReturn($autoCommitCommand);
 
-        $this->onError($commandDispatch);
+        $commandDispatch->getException()->willReturn(new \Exception());
+
+        $this->onFinalize($commandDispatch);
     }
 
     function it_does_not_perform_a_rollback_when_it_is_not_in_transaction(EventStore $eventStore, CommandDispatch $commandDispatch)
     {
         $eventStore->rollback()->shouldNotBeCalled();
 
-        $this->onError($commandDispatch);
+        $commandDispatch->getException()->willReturn(new \Exception());
+
+        $this->onFinalize($commandDispatch);
     }
 
     function it_commits_the_transaction_on_finalize(EventStore $eventStore, CommandDispatch $commandDispatch)
@@ -99,6 +104,7 @@ function it_commits_the_transaction_on_finalize(EventStore $eventStore, CommandD
         $this->onInitialize($commandDispatch);
 
         $eventStore->commit()->shouldBeCalled();
+        $commandDispatch->getException()->shouldBeCalled();
 
         $this->onFinalize($commandDispatch);
     }
@@ -109,6 +115,7 @@ function it_does_not_commit_transaction_when_command_implements_auto_commit_comm
 
         $this->onInitialize($commandDispatch);
 
+        $commandDispatch->getException()->shouldBeCalled();
         $eventStore->commit()->shouldNotBeCalled();
 
         $commandDispatch->getCommand()->willReturn($autoCommitCommand);
@@ -118,6 +125,7 @@ function it_does_not_commit_transaction_when_command_implements_auto_commit_comm
 
     function it_does_not_commit_transaction_when_it_is_not_in_transaction(EventStore $eventStore, CommandDispatch $commandDispatch)
     {
+        $commandDispatch->getException()->shouldBeCalled();
         $eventStore->commit()->shouldNotBeCalled();
 
         $this->onFinalize($commandDispatch);

src/EventStore/TransactionManager.php

@@ -78,8 +78,8 @@ public function attach(ActionEventDispatcher $events)
     {
         //Attach with a low priority, so that a potential message translator has done its job
         $this->trackHandler($events->attachListener(CommandDispatch::INITIALIZE, [$this, 'onInitialize'], -1000));
-        $this->trackHandler($events->attachListener(CommandDispatch::HANDLE_ERROR, [$this, 'onError']));
-        $this->trackHandler($events->attachListener(CommandDispatch::FINALIZE, [$this, 'onFinalize']));
+        //Attach with a high priority to rollback transaction early in case of an error
+        $this->trackHandler($events->attachListener(CommandDispatch::FINALIZE, [$this, 'onFinalize'], 1000));
     }
 
     /**
@@ -143,6 +143,11 @@ public function onError(CommandDispatch $commandDispatch)
 
     public function onFinalize(CommandDispatch $commandDispatch)
     {
+        if ($commandDispatch->getException() !== null) {
+            $this->onError($commandDispatch);
+            return;
+        }
+
         if (! $commandDispatch->getCommand() instanceof Command || $commandDispatch->getCommand() instanceof AutoCommitCommand) return;
         if (! $this->inTransaction) return;