Merge a21a83d into 3807ead

docs/tutorial/bonus_III.md

@@ -0,0 +1 @@
+# Bonus III - Decoupled Functional Core

docs/tutorial/part_VII.md

@@ -227,7 +227,7 @@ Try to check *John* in again, while keeping an eye on the monitoring app `http:/
 ## The End
 
 Congratulations! You've mastered the Event Machine tutorial. There are two bonus parts available to learn more
-about **custom projections** and **testing with Event Machine**.
+about **custom projections**, **testing with Event Machine** and how to achieve a "decoupled functional core".
 The current implementation is available as a **demo** branch of `proophsoftware/event-machine-skeleton`.
 There is a second branch called **demo-oop** available that contains a similar
 implementation, but the `Building` aggregate is designed using an object oriented approach rather than

src/Aggregate/ClosureAggregateTranslator.php

@@ -37,10 +37,13 @@ final class ClosureAggregateTranslator implements EventStoreAggregateTranslator
 
     private $eventApplyMap;
 
-    public function __construct(string $aggregateId, array $eventApplyMap)
+    private $eventClassMap;
+
+    public function __construct(string $aggregateId, array $eventApplyMap, array $eventClassMap)
     {
         $this->aggregateId = $aggregateId;
         $this->eventApplyMap = $eventApplyMap;
+        $this->eventClassMap = $eventClassMap;
     }
 
     /**
@@ -80,8 +83,9 @@ public function reconstituteAggregateFromHistory(AggregateType $aggregateType, I
         if (null === $this->aggregateReconstructor) {
             $arId = $this->aggregateId;
             $eventApplyMap = $this->eventApplyMap;
-            $this->aggregateReconstructor = function ($historyEvents) use ($arId, $aggregateType, $eventApplyMap) {
-                return static::reconstituteFromHistory($arId, $aggregateType, $eventApplyMap, $historyEvents);
+            $eventClassMap = $this->eventClassMap;
+            $this->aggregateReconstructor = function ($historyEvents) use ($arId, $aggregateType, $eventApplyMap, $eventClassMap) {
+                return static::reconstituteFromHistory($arId, $aggregateType, $eventApplyMap, $eventClassMap, $historyEvents);
             };
         }
 

src/Aggregate/GenericAggregateRoot.php

@@ -11,7 +11,7 @@
 
 namespace Prooph\EventMachine\Aggregate;
 
-use Prooph\EventMachine\Eventing\GenericJsonSchemaEvent;
+use Prooph\EventMachine\Messaging\GenericJsonSchemaEvent;
 use Prooph\EventSourcing\Aggregate\AggregateType;
 use Prooph\EventSourcing\Aggregate\AggregateTypeProvider;
 use Prooph\EventSourcing\Aggregate\Exception\RuntimeException;
@@ -36,6 +36,11 @@ final class GenericAggregateRoot implements AggregateTypeProvider
      */
     private $eventApplyMap;
 
+    /**
+     * @var string[]
+     */
+    private $eventClassMap;
+
     /**
      * @var mixed
      */
@@ -58,19 +63,20 @@ final class GenericAggregateRoot implements AggregateTypeProvider
     /**
      * @throws RuntimeException
      */
-    protected static function reconstituteFromHistory(string $aggregateId, AggregateType $aggregateType, array $eventApplyMap, \Iterator $historyEvents): self
+    protected static function reconstituteFromHistory(string $aggregateId, AggregateType $aggregateType, array $eventApplyMap, array $eventClassMap, \Iterator $historyEvents): self
     {
-        $instance = new self($aggregateId, $aggregateType, $eventApplyMap);
+        $instance = new self($aggregateId, $aggregateType, $eventApplyMap, $eventClassMap);
         $instance->replay($historyEvents);
 
         return $instance;
     }
 
-    public function __construct(string  $aggregateId, AggregateType $aggregateType, array $eventApplyMap)
+    public function __construct(string  $aggregateId, AggregateType $aggregateType, array $eventApplyMap, array $eventClassMap)
     {
         $this->aggregateId = $aggregateId;
         $this->aggregateType = $aggregateType;
         $this->eventApplyMap = $eventApplyMap;
+        $this->eventClassMap = $eventClassMap;
     }
 
     /**
@@ -134,6 +140,19 @@ private function apply(GenericJsonSchemaEvent $event): void
     {
         $apply = $this->eventApplyMap[$event->messageName()];
 
+        if (array_key_exists($event->messageName(), $this->eventClassMap)) {
+            $eventClass = $this->eventClassMap[$event->messageName()];
+
+            if (! is_callable([$eventClass, 'fromArray'])) {
+                throw new \RuntimeException(sprintf(
+                    'Custom event class %s should have a static fromArray method',
+                    $eventClass
+                ));
+            }
+
+            $event = ([$eventClass, 'fromArray'])($event->toArray());
+        }
+
         if ($this->aggregateState === null) {
             $newArState = $apply($event);
         } else {

src/Commanding/CommandProcessor.php

@@ -16,7 +16,7 @@
 use Prooph\EventMachine\Aggregate\ContextProvider;
 use Prooph\EventMachine\Aggregate\Exception\AggregateNotFound;
 use Prooph\EventMachine\Aggregate\GenericAggregateRoot;
-use Prooph\EventMachine\Eventing\GenericJsonSchemaEvent;
+use Prooph\EventMachine\Messaging\GenericJsonSchemaEvent;
 use Prooph\EventSourcing\Aggregate\AggregateRepository;
 use Prooph\EventSourcing\Aggregate\AggregateType;
 use Prooph\EventStore\EventStore;
@@ -30,6 +30,11 @@ final class CommandProcessor
      */
     private $commandName;
 
+    /**
+     * @var string|null
+     */
+    private $commandClass;
+
     /**
      * @var string
      */
@@ -60,6 +65,11 @@ final class CommandProcessor
      */
     private $eventApplyMap;
 
+    /**
+     * @var array
+     */
+    private $eventClassMap;
+
     /**
      * @var string
      */
@@ -141,7 +151,9 @@ public static function fromDescriptionArrayAndDependencies(
             $messageFactory,
             $eventStore,
             $snapshotStore,
-            $contextProvider
+            $contextProvider,
+            $description['commandClass'] ?? null,
+            $description['eventClassMap'] ?? []
         );
     }
 
@@ -157,7 +169,9 @@ public function __construct(
         MessageFactory $messageFactory,
         EventStore $eventStore,
         SnapshotStore $snapshotStore = null,
-        ContextProvider $contextProvider = null
+        ContextProvider $contextProvider = null,
+        string $commandClass = null,
+        array $eventClassMap = []
     ) {
         $this->commandName = $commandName;
         $this->aggregateType = $aggregateType;
@@ -171,6 +185,8 @@ public function __construct(
         $this->eventStore = $eventStore;
         $this->snapshotStore = $snapshotStore;
         $this->contextProvider = $contextProvider;
+        $this->commandClass = $commandClass;
+        $this->eventClassMap = $eventClassMap;
     }
 
     public function __invoke(GenericJsonSchemaCommand $command)
@@ -194,9 +210,18 @@ public function __invoke(GenericJsonSchemaCommand $command)
         $arId = (string) $payload[$this->aggregateIdentifier];
         $arRepository = $this->getAggregateRepository($arId);
         $arFuncArgs = [];
+        $commandUuid = $command->uuid()->toString();
+
+        if ($this->commandClass) {
+            if (! is_callable([$this->commandClass, 'fromArray'])) {
+                throw new \RuntimeException(sprintf('Custom command class %s should have a static fromArray method', $this->commandClass));
+            }
+
+            $command = ([$this->commandClass, 'fromArray'])($command->toArray());
+        }
 
         if ($this->createAggregate) {
-            $aggregate = new GenericAggregateRoot($arId, AggregateType::fromString($this->aggregateType), $this->eventApplyMap);
+            $aggregate = new GenericAggregateRoot($arId, AggregateType::fromString($this->aggregateType), $this->eventApplyMap, $this->eventClassMap);
             $arFuncArgs[] = $command;
         } else {
             /** @var GenericAggregateRoot $aggregate */
@@ -231,16 +256,38 @@ public function __invoke(GenericJsonSchemaCommand $command)
             }
 
             if (! is_array($event) || ! array_key_exists(0, $event) || ! array_key_exists(1, $event)
-                || ! is_string($event[0]) || ! is_array($event[1])) {
+                || ! is_string($event[0])
+                || (! is_array($event[1]) && ! is_object($event[1]))) {
                 throw new \RuntimeException(sprintf(
-                    'Event returned by aggregate of type %s while handling command %s does not has the format [string eventName, array payload]!',
+                    'Event returned by aggregate of type %s while handling command %s does not have the format [string eventName, array payload | object event]!',
                     $this->aggregateType,
                     $this->commandName
                 ));
             }
+
+            $customEvent = null;
+
             [$eventName, $payload] = $event;
 
-            $metadata = [];
+            if (is_array($payload)) {
+                $metadata = [];
+            } else {
+                //Custom event class used instead of payload array
+                if (! method_exists($payload, 'toArray')) {
+                    throw new \RuntimeException(sprintf(
+                        'Event %s returned by aggregate of type %s while handling command %s should have a toArray method',
+                        get_class($payload),
+                        $this->aggregateType,
+                        $this->commandName
+                    ));
+                }
+
+                $evtArr = $payload->toArray();
+
+                $payload = $evtArr['payload'] ?? $evtArr;
+
+                $metadata = $evtArr['metadata'] ?? [];
+            }
 
             if (array_key_exists(2, $event)) {
                 $metadata = $event[2];
@@ -258,7 +305,7 @@ public function __invoke(GenericJsonSchemaCommand $command)
             $event = $this->messageFactory->createMessageFromArray($eventName, [
                 'payload' => $payload,
                 'metadata' => array_merge([
-                    '_causation_id' => $command->uuid()->toString(),
+                    '_causation_id' => $commandUuid,
                     '_causation_name' => $this->commandName,
                 ], $metadata),
             ]);
@@ -275,7 +322,7 @@ private function getAggregateRepository(string $aggregateId): AggregateRepositor
             $this->aggregateRepository = new AggregateRepository(
                 $this->eventStore,
                 AggregateType::fromString($this->aggregateType),
-                new ClosureAggregateTranslator($aggregateId, $this->eventApplyMap),
+                new ClosureAggregateTranslator($aggregateId, $this->eventApplyMap, $this->eventClassMap),
                 $this->snapshotStore,
                 new StreamName($this->streamName)
             );

src/Commanding/CommandProcessorDescription.php

@@ -11,8 +11,8 @@
 
 namespace Prooph\EventMachine\Commanding;
 
-use Prooph\EventMachine\Eventing\EventRecorderDescription;
 use Prooph\EventMachine\EventMachine;
+use Prooph\EventMachine\Messaging\EventRecorderDescription;
 
 final class CommandProcessorDescription
 {
@@ -26,6 +26,11 @@ final class CommandProcessorDescription
      */
     private $commandName;
 
+    /**
+     * @var string|null
+     */
+    private $commandClass;
+
     /**
      * @var bool
      */
@@ -53,10 +58,11 @@ final class CommandProcessorDescription
      */
     private $contextProvider;
 
-    public function __construct(string $commandName, EventMachine $eventMachine)
+    public function __construct(string $commandName, EventMachine $eventMachine, string $commandClass = null)
     {
         $this->commandName = $commandName;
         $this->eventMachine = $eventMachine;
+        $this->commandClass = $commandClass;
     }
 
     public function withNew(string $aggregateType): self
@@ -147,6 +153,7 @@ public function __invoke(): array
 
         return [
             'commandName' => $this->commandName,
+            'commandClass' => $this->commandClass,
             'createAggregate' => $this->createAggregate,
             'aggregateType' => $this->aggregateType,
             'aggregateIdentifier' => $this->aggregateIdentifier,

src/Commanding/CommandToProcessorRouter.php

@@ -33,6 +33,11 @@ final class CommandToProcessorRouter extends AbstractPlugin
      */
     private $aggregateDescriptions;
 
+    /**
+     * @var array
+     */
+    private $eventClassMap;
+
     /**
      * @var MessageFactory
      */
@@ -56,13 +61,15 @@ final class CommandToProcessorRouter extends AbstractPlugin
     public function __construct(
         array $routingMap,
         array $aggregateDescriptions,
+        array $eventClassMap,
         MessageFactory $messageFactory,
         EventStore $eventStore,
         ContextProviderFactory $providerFactory,
         SnapshotStore $snapshotStore = null
     ) {
         $this->routingMap = $routingMap;
         $this->aggregateDescriptions = $aggregateDescriptions;
+        $this->eventClassMap = $eventClassMap;
         $this->messageFactory = $messageFactory;
         $this->eventStore = $eventStore;
         $this->contextProviderFactory = $providerFactory;
@@ -103,6 +110,7 @@ public function onRouteMessage(ActionEvent $actionEvent): void
         }
 
         $processorDesc['eventApplyMap'] = $aggregateDesc['eventApplyMap'];
+        $processorDesc['eventClassMap'] = $this->eventClassMap;
 
         $contextProvider = $processorDesc['contextProvider'] ? $this->contextProviderFactory->build($processorDesc['contextProvider']) : null;