[Messenger] Add StackInterface, allowing to unstack the call stack

Author: nicolas-grekas

CHANGELOG.md

@@ -7,7 +7,7 @@ CHANGELOG
  * The component is not experimental anymore
  * All the changes below are BC BREAKS
  * `MessageBusInterface::dispatch()` and `MiddlewareInterface::handle()` now return `void`
- * `MiddlewareInterface::handle()` now require an `Envelope` as first argument
+ * `MiddlewareInterface::handle()` now require an `Envelope` as first argument and a `StackInterface` as second
  * `EnvelopeAwareInterface` has been removed
  * The signature of `Amqp*` classes changed to take a `Connection` as a first argument and an optional
    `Serializer` as a second argument.
@@ -16,7 +16,6 @@ CHANGELOG
    Instead, it accepts the sender instance itself instead of its identifier in the container.
  * `MessageSubscriberInterface::getHandledMessages()` return value has changed. The value of an array item
    needs to be an associative array or the method name.
- * `ValidationMiddleware::handle()` and `SendMessageMiddleware::handle()` now require an `Envelope` object
  * `StampInterface` replaces `EnvelopeItemInterface` and doesn't extend `Serializable` anymore
  * The `ConsumeMessagesCommand` class now takes an instance of `Psr\Container\ContainerInterface`
    as first constructor argument

MessageBus.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger;
 
 use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackMiddleware;
 
 /**
  * @author Samuel Roze <samuel.roze@gmail.com>
@@ -64,14 +65,8 @@ public function dispatch($message): void
         if (!$middlewareIterator->valid()) {
             return;
         }
-        $next = static function (Envelope $envelope) use ($middlewareIterator, &$next) {
-            $middlewareIterator->next();
+        $stack = new StackMiddleware($middlewareIterator);
 
-            if ($middlewareIterator->valid()) {
-                $middlewareIterator->current()->handle($envelope, $next);
-            }
-        };
-
-        $middlewareIterator->current()->handle($message instanceof Envelope ? $message : new Envelope($message), $next);
+        $middlewareIterator->current()->handle($message instanceof Envelope ? $message : new Envelope($message), $stack);
     }
 }

Middleware/ActivationMiddleware.php

@@ -35,12 +35,12 @@ public function __construct(MiddlewareInterface $inner, $activated)
     /**
      * {@inheritdoc}
      */
-    public function handle(Envelope $envelope, callable $next): void
+    public function handle(Envelope $envelope, StackInterface $stack): void
     {
         if (\is_callable($this->activated) ? ($this->activated)($envelope) : $this->activated) {
-            $this->inner->handle($envelope, $next);
+            $this->inner->handle($envelope, $stack);
         } else {
-            $next($envelope);
+            $stack->next()->handle($envelope, $stack);
         }
     }
 }

Middleware/HandleMessageMiddleware.php

@@ -34,11 +34,11 @@ public function __construct(HandlerLocatorInterface $messageHandlerLocator, bool
      *
      * @throws NoHandlerForMessageException When no handler is found and $allowNoHandlers is false
      */
-    public function handle(Envelope $envelope, callable $next): void
+    public function handle(Envelope $envelope, StackInterface $stack): void
     {
         if (null !== $handler = $this->messageHandlerLocator->getHandler($envelope)) {
             $handler($envelope->getMessage());
-            $next($envelope);
+            $stack->next()->handle($envelope, $stack);
         } elseif (!$this->allowNoHandlers) {
             throw new NoHandlerForMessageException(sprintf('No handler for message "%s".', \get_class($envelope->getMessage())));
         }

Middleware/LoggingMiddleware.php

@@ -29,7 +29,7 @@ public function __construct(LoggerInterface $logger)
     /**
      * {@inheritdoc}
      */
-    public function handle(Envelope $envelope, callable $next): void
+    public function handle(Envelope $envelope, StackInterface $stack): void
     {
         $message = $envelope->getMessage();
         $context = array(
@@ -39,7 +39,7 @@ public function handle(Envelope $envelope, callable $next): void
         $this->logger->debug('Starting handling message {name}', $context);
 
         try {
-            $next($envelope);
+            $stack->next()->handle($envelope, $stack);
         } catch (\Throwable $e) {
             $context['exception'] = $e;
             $this->logger->warning('An exception occurred while handling message {name}', $context);

Middleware/MiddlewareInterface.php

@@ -18,16 +18,5 @@
  */
 interface MiddlewareInterface
 {
-    /**
-     * @param callable|NextInterface $next
-     */
-    public function handle(Envelope $envelope, callable $next): void;
-}
-
-/**
- * @internal
- */
-interface NextInterface
-{
-    public function __invoke(Envelope $envelope): void;
+    public function handle(Envelope $envelope, StackInterface $stack): void;
 }

Middleware/SendMessageMiddleware.php

@@ -34,11 +34,11 @@ public function __construct(SenderLocatorInterface $senderLocator, array $messag
     /**
      * {@inheritdoc}
      */
-    public function handle(Envelope $envelope, callable $next): void
+    public function handle(Envelope $envelope, StackInterface $stack): void
     {
         if ($envelope->get(ReceivedStamp::class)) {
             // It's a received message. Do not send it back:
-            $next($envelope);
+            $stack->next()->handle($envelope, $stack);
 
             return;
         }
@@ -54,6 +54,6 @@ public function handle(Envelope $envelope, callable $next): void
             }
         }
 
-        $next($envelope);
+        $stack->next()->handle($envelope, $stack);
     }
 }

Middleware/StackInterface.php

@@ -0,0 +1,23 @@
+<?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;
+
+/**
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+interface StackInterface
+{
+    /**
+     * Returns the next middleware to process a message.
+     */
+    public function next(): MiddlewareInterface;
+}

Middleware/StackMiddleware.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\Middleware;
+
+use Symfony\Component\Messenger\Envelope;
+
+/**
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+class StackMiddleware implements MiddlewareInterface, StackInterface
+{
+    private $middlewareIterator;
+
+    public function __construct(\Iterator $middlewareIterator = null)
+    {
+        $this->middlewareIterator = $middlewareIterator;
+    }
+
+    public function next(): MiddlewareInterface
+    {
+        if (null === $iterator = $this->middlewareIterator) {
+            return $this;
+        }
+        $iterator->next();
+
+        if (!$iterator->valid()) {
+            $this->middlewareIterator = null;
+
+            return $this;
+        }
+
+        return $iterator->current();
+    }
+
+    public function handle(Envelope $envelope, StackInterface $stack): void
+    {
+        // no-op: this is the last null middleware
+    }
+}

Middleware/TraceableMiddleware.php

@@ -37,7 +37,7 @@ public function __construct(MiddlewareInterface $inner, Stopwatch $stopwatch, st
     /**
      * {@inheritdoc}
      */
-    public function handle(Envelope $envelope, callable $next): void
+    public function handle(Envelope $envelope, StackInterface $stack): void
     {
         $class = \get_class($this->inner);
         $eventName = 'c' === $class[0] && 0 === strpos($class, "class@anonymous\0") ? get_parent_class($class).'@anonymous' : $class;
@@ -49,15 +49,52 @@ public function handle(Envelope $envelope, callable $next): void
         $this->stopwatch->start($eventName, $this->eventCategory);
 
         try {
-            $this->inner->handle($envelope, function (Envelope $envelope) use ($next, $eventName) {
-                $this->stopwatch->stop($eventName);
-                $next($envelope);
-                $this->stopwatch->start($eventName, $this->eventCategory);
-            });
+            $this->inner->handle($envelope, new TraceableInnerMiddleware($stack, $this->stopwatch, $eventName, $this->eventCategory));
         } finally {
             if ($this->stopwatch->isStarted($eventName)) {
                 $this->stopwatch->stop($eventName);
             }
         }
     }
 }
+
+/**
+ * @internal
+ */
+class TraceableInnerMiddleware implements MiddlewareInterface, StackInterface
+{
+    private $stack;
+    private $stopwatch;
+    private $eventName;
+    private $eventCategory;
+
+    public function __construct(StackInterface $stack, Stopwatch $stopwatch, string $eventName, string $eventCategory)
+    {
+        $this->stack = $stack;
+        $this->stopwatch = $stopwatch;
+        $this->eventName = $eventName;
+        $this->eventCategory = $eventCategory;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function handle(Envelope $envelope, StackInterface $stack): void
+    {
+        $this->stopwatch->stop($this->eventName);
+        if ($this === $stack) {
+            $this->stack->next()->handle($envelope, $this->stack);
+        } else {
+            $stack->next()->handle($envelope, $stack);
+        }
+        $this->stopwatch->start($this->eventName, $this->eventCategory);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function next(): MiddlewareInterface
+    {
+        return $this;
+    }
+}