Merge pull request #63 from Lctrs/messenger-middleware

Add EventStoreTransactionMiddleware to be used with symfony/messenger

composer.json

@@ -69,7 +69,8 @@
     "friendsofphp/php-cs-fixer": "^2.8.1",
     "prooph/php-cs-fixer-config": "^0.2.1",
     "matthiasnoback/symfony-dependency-injection-test": "^2.3",
-    "phpstan/phpstan": "^0.9.2"
+    "phpstan/phpstan": "^0.9.2",
+    "symfony/messenger": "^4.3"
   },
   "suggest": {
     "prooph/event-store-bus-bridge": "To Marry CQRS (ProophSerivceBus) with Event Sourcing"

doc/bookdown.json

@@ -6,7 +6,8 @@
     {"event_store": "event_store.md"},
     {"projection_manager": "projection_manager.md"},
     {"event_store_bus_bridge": "event_store_bus_bridge.md"},
-    {"configuration_reference": "configuration_reference.md"}
+    {"configuration_reference": "configuration_reference.md"},
+    {"messenger": "messenger.md"}
   ],
   "target": "./html",
   "tocDepth": 2,

doc/messenger.md

@@ -0,0 +1,24 @@
+# Messenger integration
+
+This bundle provides a middleware for the
+`symfony/messenger` component (from version `4.3`) which handles
+starting/committing/rolling back a transaction when sending a command
+to the bus.
+
+Here is an example configuration on how to use it:
+```yaml
+# app/config/messenger.yaml
+
+framework:
+    messenger:
+        buses:
+            command.bus:
+                middleware:
+                    - my_eventstore_transaction_middleware
+
+services:
+    my_eventstore_transaction_middleware:
+        class: Prooph\Bundle\EventStore\Messenger\EventStoreTransactionMiddleware
+        arguments:
+            - '@my_transactional_event_store'
+```

src/Messenger/EventStoreTransactionMiddleware.php

@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Prooph\Bundle\EventStore\Messenger;
+
+use Prooph\EventStore\TransactionalEventStore;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+use Symfony\Component\Messenger\Stamp\HandledStamp;
+use Throwable;
+
+final class EventStoreTransactionMiddleware implements MiddlewareInterface
+{
+    /** @var TransactionalEventStore */
+    private $eventStore;
+
+    public function __construct(TransactionalEventStore $eventStore)
+    {
+        $this->eventStore = $eventStore;
+    }
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        $this->eventStore->beginTransaction();
+
+        try {
+            $envelope = $stack->next()->handle($envelope, $stack);
+
+            $this->eventStore->commit();
+        } catch (Throwable $e) {
+            $this->eventStore->rollback();
+
+            if ($e instanceof HandlerFailedException) {
+                // Remove all HandledStamp from the envelope so the retry will execute all handlers again.
+                // When a handler fails, the queries of allegedly successful previous handlers just got rolled back.
+                throw new HandlerFailedException(
+                    $e->getEnvelope()->withoutAll(HandledStamp::class),
+                    $e->getNestedExceptions()
+                );
+            }
+
+            throw $e;
+        }
+
+        return $envelope;
+    }
+}

test/Messenger/EventStoreTransactionMiddlewareTest.php

@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ProophTest\Bundle\EventStore\Messenger;
+
+use LogicException;
+use Prooph\Bundle\EventStore\Messenger\EventStoreTransactionMiddleware;
+use Prooph\EventStore\TransactionalEventStore;
+use stdClass;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Stamp\HandledStamp;
+use Symfony\Component\Messenger\Test\Middleware\MiddlewareTestCase;
+use Throwable;
+
+class EventStoreTransactionMiddlewareTest extends MiddlewareTestCase
+{
+    /** @var TransactionalEventStore */
+    private $eventStore;
+    /** @var EventStoreTransactionMiddleware */
+    private $middleware;
+
+    public function setUp(): void
+    {
+        $this->eventStore = $this->createMock(TransactionalEventStore::class);
+        $this->middleware = new EventStoreTransactionMiddleware($this->eventStore);
+    }
+
+    public function testMiddlewareWrapsInTransactionAndFlushes(): void
+    {
+        $this->eventStore->expects($this->once())
+            ->method('beginTransaction');
+        $this->eventStore->expects($this->once())
+            ->method('commit');
+
+        $this->middleware->handle(new Envelope(new stdClass()), $this->getStackMock());
+    }
+
+    /**
+     * @expectedException \RuntimeException
+     * @expectedExceptionMessage Thrown from next middleware.
+     */
+    public function testTransactionIsRolledBackOnException(): void
+    {
+        $this->eventStore->expects($this->once())
+            ->method('beginTransaction');
+        $this->eventStore->expects($this->once())
+            ->method('rollback');
+
+        $this->middleware->handle(new Envelope(new stdClass()), $this->getThrowingStackMock());
+    }
+
+    public function testItResetsHandledStampsOnHandlerFailedException(): void
+    {
+        $this->eventStore->expects($this->once())
+            ->method('beginTransaction');
+        $this->eventStore->expects($this->once())
+            ->method('rollback');
+
+        $envelop = new Envelope(new stdClass(), [
+            new HandledStamp('dummy', 'dummy'),
+        ]);
+
+        $exception = null;
+        try {
+            $this->middleware->handle($envelop, $this->getThrowingStackMock(new HandlerFailedException($envelop, [
+                new LogicException('dummy exception'),
+            ])));
+        } catch (Throwable $e) {
+            $exception = $e;
+        }
+
+        $this->assertInstanceOf(HandlerFailedException::class, $exception);
+        /** @var HandlerFailedException $exception */
+        $this->assertSame([], $exception->getEnvelope()->all(HandledStamp::class));
+    }
+}