symfony · sroze · Mar 19, 2019 · Oct 13, 2018 · weaverryan · Mar 14, 2019
diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
@@ -1608,7 +1608,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
        }

        $defaultMiddleware = [
-            'before' => [],
+            'before' => [['id' => 'dispatch_after_current_bus']],
            'after' => [['id' => 'send_message'], ['id' => 'handle_message']],
        ];
        foreach ($config['buses'] as $busId => $bus) {

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml b/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.xml
@@ -39,6 +39,8 @@
            </call>
        </service>

+        <service id="messenger.middleware.dispatch_after_current_bus" class="Symfony\Component\Messenger\Middleware\DispatchAfterCurrentBusMiddleware" />
+
        <service id="messenger.middleware.validation" class="Symfony\Component\Messenger\Middleware\ValidationMiddleware">
            <argument type="service" id="validator" />
        </service>

diff --git a/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php b/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php
@@ -702,12 +702,14 @@ public function testMessengerWithMultipleBuses()
        $this->assertTrue($container->has('messenger.bus.commands'));
        $this->assertSame([], $container->getDefinition('messenger.bus.commands')->getArgument(0));
        $this->assertEquals([
+            ['id' => 'dispatch_after_current_bus'],
            ['id' => 'send_message'],
            ['id' => 'handle_message'],
        ], $container->getParameter('messenger.bus.commands.middleware'));
        $this->assertTrue($container->has('messenger.bus.events'));
        $this->assertSame([], $container->getDefinition('messenger.bus.events')->getArgument(0));
        $this->assertEquals([
+            ['id' => 'dispatch_after_current_bus'],
            ['id' => 'with_factory', 'arguments' => ['foo', true, ['bar' => 'baz']]],
            ['id' => 'send_message'],
            ['id' => 'handle_message'],

diff --git a/src/Symfony/Bundle/FrameworkBundle/composer.json b/src/Symfony/Bundle/FrameworkBundle/composer.json
@@ -43,7 +43,7 @@
        "symfony/form": "^4.3",
        "symfony/expression-language": "~3.4|~4.0",
        "symfony/http-client": "^4.3",
-        "symfony/messenger": "^4.2",
+        "symfony/messenger": "^4.3",
        "symfony/mime": "^4.3",
        "symfony/process": "~3.4|~4.0",
        "symfony/security-core": "~3.4|~4.0",

diff --git a/src/Symfony/Component/Messenger/Exception/DelayedMessageHandlingException.php b/src/Symfony/Component/Messenger/Exception/DelayedMessageHandlingException.php
@@ -0,0 +1,48 @@
+<?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 queued messages from {@link DispatchAfterCurrentBusMiddleware},
+ * some handlers caused an exception. This exception contains all those handler exceptions.
+ *
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ */
+class DelayedMessageHandlingException extends \RuntimeException implements ExceptionInterface
+{
+    private $exceptions;
+
+    public function __construct(array $exceptions)
+    {
+        $exceptionMessages = implode(", \n", array_map(
+            function (\Throwable $e) {
+                return \get_class($e).': '.$e->getMessage();
+            },
+            $exceptions
+        ));
+
+        if (1 === \count($exceptions)) {
+            $message = sprintf("A delayed message handler threw an exception: \n\n%s", $exceptionMessages);
+        } else {
+            $message = sprintf("Some delayed message handlers threw an exception: \n\n%s", $exceptionMessages);
+        }
+
+        $this->exceptions = $exceptions;
+
+        parent::__construct($message, 0, $exceptions[0]);
+    }
+
+    public function getExceptions(): array
+    {
+        return $this->exceptions;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Middleware/DispatchAfterCurrentBusMiddleware.php b/src/Symfony/Component/Messenger/Middleware/DispatchAfterCurrentBusMiddleware.php
@@ -0,0 +1,128 @@
+<?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\Envelope;
+use Symfony\Component\Messenger\Exception\DelayedMessageHandlingException;
+use Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp;
+
+/**
+ * Allow to configure messages to be handled after the current bus is finished.
+ *
+ * I.e, messages dispatched from a handler with a DispatchAfterCurrentBus stamp
+ * will actually be handled once the current message being dispatched is fully
+ * handled.
+ *
+ * For instance, using this middleware before the DoctrineTransactionMiddleware
+ * means sub-dispatched messages with a DispatchAfterCurrentBus stamp would be
+ * handled after the Doctrine transaction has been committed.
+ *
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ */
+class DispatchAfterCurrentBusMiddleware implements MiddlewareInterface
+{
+    /**
+     * @var QueuedEnvelope[] A queue of messages and next middleware
+     */
+    private $queue = [];
+
+    /**
+     * @var bool this property is used to signal if we are inside a the first/root call to
+     *           MessageBusInterface::dispatch() or if dispatch has been called inside a message handler
+     */
+    private $isRootDispatchCallRunning = false;
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        if (null !== $envelope->last(DispatchAfterCurrentBusStamp::class)) {
+            if (!$this->isRootDispatchCallRunning) {
+                throw new \LogicException(sprintf('You can only use a "%s" stamp in the context of a message handler.', DispatchAfterCurrentBusStamp::class));
+            }
+            $this->queue[] = new QueuedEnvelope($envelope, $stack);
+
+            return $envelope;
+        }
+
+        if ($this->isRootDispatchCallRunning) {
+            /*
+             * A call to MessageBusInterface::dispatch() was made from inside the main bus handling,
+             * but the message does not have the stamp. So, process it like normal.
+             */
+            return $stack->next()->handle($envelope, $stack);
+        }
+
+        // First time we get here, mark as inside a "root dispatch" call:
+        $this->isRootDispatchCallRunning = true;
+        try {
+            // Execute the whole middleware stack & message handling for main dispatch:
+            $returnedEnvelope = $stack->next()->handle($envelope, $stack);
+        } catch (\Throwable $exception) {
+            /*
+             * Whenever an exception occurs while handling a message that has
+             * queued other messages, we drop the queued ones.
+             * This is intentional since the queued commands were likely dependent
+             * on the preceding command.
+             */
+            $this->queue = [];
+            $this->isRootDispatchCallRunning = false;
+
+            throw $exception;
+        }
+
+        // "Root dispatch" call is finished, dispatch stored messages.
+        $exceptions = [];
+        while (null !== $queueItem = array_shift($this->queue)) {
+            try {
+                // Execute the stored messages
+                $queueItem->getStack()->next()->handle($queueItem->getEnvelope(), $queueItem->getStack());
+            } catch (\Exception $exception) {
+                // Gather all exceptions
+                $exceptions[] = $exception;
+            }
+        }
+
+        $this->isRootDispatchCallRunning = false;
+        if (\count($exceptions) > 0) {
+            throw new DelayedMessageHandlingException($exceptions);
+        }
+
+        return $returnedEnvelope;
+    }
+}
+
+/**
+ * @internal
+ */
+final class QueuedEnvelope
+{
+    /** @var Envelope */
+    private $envelope;
+
+    /** @var StackInterface */
+    private $stack;
+
+    public function __construct(Envelope $envelope, StackInterface $stack)
+    {
+        $this->envelope = $envelope;
+        $this->stack = $stack;
+    }
+
+    public function getEnvelope(): Envelope
+    {
+        return $this->envelope;
+    }
+
+    public function getStack(): StackInterface
+    {
+        return $this->stack;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Stamp/DispatchAfterCurrentBusStamp.php b/src/Symfony/Component/Messenger/Stamp/DispatchAfterCurrentBusStamp.php
@@ -0,0 +1,25 @@
+<?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.
+ */
+
+declare(strict_types=1);
+
+namespace Symfony\Component\Messenger\Stamp;
+
+/**
+ * Marker item to tell this message should be handled in after the current bus has finished.
+ *
+ * @see \Symfony\Component\Messenger\Middleware\DispatchAfterCurrentBusMiddleware
+ *
+ * @author Tobias Nyholm <tobias.nyholm@gmail.com>
+ */
+class DispatchAfterCurrentBusStamp implements StampInterface
+{
+}