Adding support for record messages

src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml

@@ -34,6 +34,11 @@
             <argument type="service" id="validator" />
         </service>
 
+        <service id="messenger.middleware.record_messages" class="Symfony\Component\Messenger\Middleware\HandleRecordedMessageMiddleware" abstract="true">
+            <argument /> <!-- Message bus -->
+            <argument type="service" id="messenger.recorder" />
+        </service>
+
         <!-- Logging -->
         <service id="messenger.middleware.logging" class="Symfony\Component\Messenger\Middleware\LoggingMiddleware" abstract="true">
             <argument type="service" id="logger" />
@@ -62,5 +67,11 @@
 
             <tag name="messenger.transport_factory" />
         </service>
+
+        <!-- Message recorder -->
+        <service id="messenger.recorder" class="Symfony\Component\Messenger\MessageRecorder">
+            <tag name="kernel.reset" method="reset" />
+        </service>
+        <service id="Symfony\Component\Messenger\MessageRecorderInterface" alias="messenger.recorder" />
     </services>
 </container>

src/Symfony/Component/Messenger/Exception/MessageHandlingException.php

@@ -0,0 +1,41 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * When handling messages, some handlers caused an exception. This exception
+ * contains all those handler exceptions.
+ *
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ */
+class MessageHandlingException extends \RuntimeException implements ExceptionInterface
+{
+    private $exceptions = array();
+
+    public function __construct(array $exceptions)
+    {
+        $message = sprintf(
+            "Some handlers for recorded messages threw an exception. Their messages were: \n\n%s",
+            implode(", \n", array_map(function (\Throwable $e) {
+                return $e->getMessage();
+            }, $exceptions))
+        );
+
+        $this->exceptions = $exceptions;
+        parent::__construct($message);
+    }
+
+    public function getExceptions(): array
+    {
+        return $this->exceptions;
+    }
+}

src/Symfony/Component/Messenger/MessageRecorder.php

@@ -0,0 +1,55 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger;
+
+use Symfony\Contracts\Service\ResetInterface;
+
+/**
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ * @author Matthias Noback <matthiasnoback@gmail.com>
+ */
+class MessageRecorder implements MessageRecorderInterface, RecordedMessageCollectionInterface, ResetInterface
+{
+    private $messages = array();
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getRecordedMessages(): array
+    {
+        return $this->messages;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function resetRecordedMessages(): void
+    {
+        $this->reset();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function reset()
+    {
+        $this->messages = array();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function record($message): void
+    {
+        $this->messages[] = $message;
+    }
+}

src/Symfony/Component/Messenger/MessageRecorderInterface.php

@@ -0,0 +1,26 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger;
+
+/**
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ * @author Matthias Noback <matthiasnoback@gmail.com>
+ */
+interface MessageRecorderInterface
+{
+    /**
+     * Record a message.
+     *
+     * @param object $message
+     */
+    public function record($message);
+}

src/Symfony/Component/Messenger/Middleware/HandleRecordedMessageMiddleware.php

@@ -0,0 +1,71 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Middleware;
+
+use Symfony\Component\Messenger\Exception\MessageHandlingException;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\RecordedMessageCollectionInterface;
+
+/**
+ * A middleware that takes all recorded messages and dispatch them to the bus.
+ *
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ * @author Matthias Noback <matthiasnoback@gmail.com>
+ */
+class HandleRecordedMessageMiddleware implements MiddlewareInterface
+{
+    private $messageRecorder;
+    private $messageBus;
+
+    public function __construct(MessageBusInterface $messageBus, RecordedMessageCollectionInterface $messageRecorder)
+    {
+        $this->messageRecorder = $messageRecorder;
+        $this->messageBus = $messageBus;
+    }
+
+    public function handle($message, callable $next)
+    {
+        // Make sure the recorder is empty before we begin
+        $this->messageRecorder->resetRecordedMessages();
+
+        try {
+            $returnData = $next($message);
+        } catch (\Throwable $exception) {
+            $this->messageRecorder->resetRecordedMessages();
+
+            throw $exception;
+        }
+
+        $exceptions = array();
+        while (!empty($recordedMessages = $this->messageRecorder->getRecordedMessages())) {
+            $this->messageRecorder->resetRecordedMessages();
+            // Assert: The message recorder is empty, all messages are in $recordedMessages
+
+            foreach ($recordedMessages as $recordedMessage) {
+                try {
+                    $this->messageBus->dispatch($recordedMessage);
+                } catch (\Throwable $exception) {
+                    $exceptions[] = $exception;
+                }
+            }
+        }
+
+        if (!empty($exceptions)) {
+            if (1 === count($exceptions)) {
+                throw $exceptions[0];
+            }
+            throw new MessageHandlingException($exceptions);
+        }
+
+        return $returnData;
+    }
+}

src/Symfony/Component/Messenger/RecordedMessageCollectionInterface.php

@@ -0,0 +1,31 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger;
+
+/**
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ * @author Matthias Noback <matthiasnoback@gmail.com>
+ */
+interface RecordedMessageCollectionInterface
+{
+    /**
+     * Fetch recorded messages.
+     *
+     * @return object[]
+     */
+    public function getRecordedMessages(): array;
+
+    /**
+     * Remove all recorded messages.
+     */
+    public function resetRecordedMessages(): void;
+}