Introduce PushMiddlewareConfig + Refactor and make internal PushMiddlewareDispatcher (#283)

Author: vjik

config/di.php

@@ -14,9 +14,9 @@
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactory;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactoryInterface;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactoryInterface;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Queue\Worker\Worker as QueueWorker;
 use Yiisoft\Queue\Worker\WorkerInterface;
 
@@ -36,8 +36,8 @@
     PushMiddlewareFactoryInterface::class => PushMiddlewareFactory::class,
     ConsumeMiddlewareFactoryInterface::class => ConsumeMiddlewareFactory::class,
     FailureMiddlewareFactoryInterface::class => FailureMiddlewareFactory::class,
-    PushMiddlewareDispatcher::class => [
-        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-push']],
+    PushMiddlewareConfig::class => [
+        '__construct()' => ['commonMiddlewareDefinitions' => $params['yiisoft/queue']['middlewares-push']],
     ],
     ConsumeMiddlewareDispatcher::class => [
         '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-consume']],

docs/guide/en/configuration-manual.md

@@ -22,8 +22,8 @@ use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareFactory;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactory;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\Worker\Worker;
 
@@ -51,7 +51,7 @@ $failureMiddlewareDispatcher = new FailureMiddlewareDispatcher(
     [],
 );
 
-$pushMiddlewareDispatcher = new PushMiddlewareDispatcher(
+$pushMiddlewareConfig = new PushMiddlewareConfig(
     new PushMiddlewareFactory($container, $callableFactory),
 );
 
@@ -75,7 +75,7 @@ $queue = new Queue(
     $worker,
     $loop,
     $logger,
-    $pushMiddlewareDispatcher,
+    $pushMiddlewareConfig,
 );
 
 // Now you can push messages

docs/guide/en/synchronous-mode.md

@@ -15,15 +15,15 @@ $logger = $DIContainer->get(\Psr\Log\LoggerInterface::class);
 
 $worker = $DIContainer->get(\Yiisoft\Queue\Worker\WorkerInterface::class);
 $loop = $DIContainer->get(\Yiisoft\Queue\Cli\LoopInterface::class);
-$pushMiddlewareDispatcher = $DIContainer->get(
-    \Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher::class
+$pushMiddlewareConfig = $DIContainer->get(
+    \Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig::class
 );
 
 $queue = new Yiisoft\Queue\Queue(
     $worker,
     $loop,
     $logger,
-    $pushMiddlewareDispatcher,
+    $pushMiddlewareConfig,
 );
 ```
 

src/Middleware/Push/PushMiddlewareConfig.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Middleware\Push;
+
+/**
+ * Holds the push middleware factory and the list of common middleware definitions
+ * applied to queues.
+ */
+final class PushMiddlewareConfig
+{
+    /**
+     * @param PushMiddlewareFactoryInterface $middlewareFactory Factory used to instantiate middleware from definitions.
+     * @param array<array|callable|PushMiddlewareInterface|string> $commonMiddlewareDefinitions Middleware definitions.
+     */
+    public function __construct(
+        public readonly PushMiddlewareFactoryInterface $middlewareFactory,
+        public readonly array $commonMiddlewareDefinitions = [],
+    ) {}
+}

src/Middleware/Push/PushMiddlewareDispatcher.php

@@ -6,7 +6,11 @@
 
 use Closure;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Queue;
 
+/**
+ * @internal Used internally by {@see Queue}.
+ */
 final class PushMiddlewareDispatcher
 {
     /**
@@ -15,35 +19,44 @@ final class PushMiddlewareDispatcher
      * @var PushMiddlewareStack|null The middleware stack.
      */
     private ?PushMiddlewareStack $stack = null;
+
     /**
-     * @var array[]|callable[]|PushMiddlewareInterface[]|string[]
+     * @param PushMiddlewareFactoryInterface $middlewareFactory Factory used to instantiate middleware.
+     * @param array<array|callable|PushMiddlewareInterface|string> $middlewareDefinitions Middleware definitions.
+     * @param PushHandlerInterface $finishHandler Finish message handler.
      */
-    private array $middlewareDefinitions;
-
     public function __construct(
         private readonly PushMiddlewareFactoryInterface $middlewareFactory,
-        array|callable|string|PushMiddlewareInterface ...$middlewareDefinitions,
-    ) {
-        $this->middlewareDefinitions = $middlewareDefinitions;
-    }
+        private array $middlewareDefinitions,
+        private PushHandlerInterface $finishHandler,
+    ) {}
 
     /**
      * Dispatch message through middleware to get response.
      *
      * @param MessageInterface $message Message to pass to middleware.
-     * @param PushHandlerInterface $finishHandler Handler to use in case no middleware produced a response.
      */
-    public function dispatch(
-        MessageInterface $message,
-        PushHandlerInterface $finishHandler,
-    ): MessageInterface {
+    public function dispatch(MessageInterface $message): MessageInterface
+    {
         if ($this->stack === null) {
-            $this->stack = new PushMiddlewareStack($this->buildMiddlewares(), $finishHandler);
+            $this->stack = new PushMiddlewareStack($this->buildMiddlewares(), $this->finishHandler);
         }
 
         return $this->stack->handlePush($message);
     }
 
+    public function withFinishHandler(PushHandlerInterface $finishHandler): self
+    {
+        $instance = clone $this;
+        $instance->finishHandler = $finishHandler;
+
+        // Fixes a memory leak.
+        unset($instance->stack);
+        $instance->stack = null;
+
+        return $instance;
+    }
+
     /**
      * Returns new instance with middleware handlers replaced with the ones provided.
      *

src/Queue.php

@@ -11,8 +11,9 @@
 use Yiisoft\Queue\Message\MessageInterface;
 use Yiisoft\Queue\Middleware\Push\AdapterPushHandler;
 use Yiisoft\Queue\Middleware\Push\PushHandlerInterface;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareInterface;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareInterface;
 use Yiisoft\Queue\Middleware\Push\SynchronousPushHandler;
 use Yiisoft\Queue\Worker\WorkerInterface;
 use Yiisoft\Queue\Message\IdEnvelope;
@@ -25,13 +26,6 @@ final class Queue implements QueueInterface
      */
     private array $middlewareDefinitions;
 
-    /**
-     * @var PushHandlerInterface The final push handler in the middleware chain, responsible
-     * for actually sending the message. Uses {@see SynchronousPushHandler} in synchronous mode or
-     * {@see AdapterPushHandler} otherwise.
-     */
-    private PushHandlerInterface $finalPushHandler;
-
     private string $name;
 
     /**
@@ -40,11 +34,18 @@ final class Queue implements QueueInterface
      */
     private PushMiddlewareDispatcher $dispatcher;
 
+    /**
+     * @var PushMiddlewareDispatcher The base dispatcher built from {@see PushMiddlewareConfig}.
+     * Holds the common middleware applied to all queues.
+     */
+    private PushMiddlewareDispatcher $baseDispatcher;
+
     /**
      * @param WorkerInterface $worker The worker that processes messages.
      * @param LoopInterface $loop The loop for controlling message processing.
      * @param LoggerInterface $logger The logger for debug and informational messages.
-     * @param PushMiddlewareDispatcher $baseDispatcher The middleware dispatcher.
+     * @param PushMiddlewareConfig $middlewareConfig The push middleware configuration: factory and common middleware
+     * definitions.
      * @param AdapterInterface|null $adapter The message adapter (`null` for synchronous mode).
      * @param string|BackedEnum $name The queue name.
      * @param PushMiddlewareInterface|callable|array|string ...$middlewareDefinitions Queue-specific middleware
@@ -54,19 +55,25 @@ public function __construct(
         private readonly WorkerInterface $worker,
         private readonly LoopInterface $loop,
         private readonly LoggerInterface $logger,
-        private readonly PushMiddlewareDispatcher $baseDispatcher,
+        PushMiddlewareConfig $middlewareConfig,
         private readonly ?AdapterInterface $adapter = null,
         string|BackedEnum $name = QueueProviderInterface::DEFAULT_QUEUE,
         PushMiddlewareInterface|callable|array|string ...$middlewareDefinitions,
     ) {
         $this->name = StringNormalizer::normalize($name);
-        $this->finalPushHandler = $this->createFinalPushHandler();
+        $this->baseDispatcher = new PushMiddlewareDispatcher(
+            $middlewareConfig->middlewareFactory,
+            $middlewareConfig->commonMiddlewareDefinitions,
+            $this->createFinalPushHandler(),
+        );
         $this->setMiddlewaresAndPrepareDispatcher($middlewareDefinitions);
     }
 
     public function __clone()
     {
-        $this->finalPushHandler = $this->createFinalPushHandler();
+        $finalPushHandler = $this->createFinalPushHandler();
+        $this->baseDispatcher = $this->baseDispatcher->withFinishHandler($finalPushHandler);
+        $this->dispatcher = $this->dispatcher->withFinishHandler($finalPushHandler);
     }
 
     public function getName(): string
@@ -81,7 +88,7 @@ public function push(MessageInterface $message): MessageInterface
             ['messageType' => $message->getType()],
         );
 
-        $message = $this->dispatcher->dispatch($message, $this->finalPushHandler);
+        $message = $this->dispatcher->dispatch($message);
 
         if ($this->isSynchronous()) {
             $this->logger->info(

tests/Benchmark/QueueBench.php

@@ -18,8 +18,8 @@
 use Yiisoft\Queue\Middleware\FailureHandling\FailureEnvelope;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactory;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Tests\Benchmark\Support\VoidAdapter;
@@ -59,7 +59,7 @@ public function __construct()
             $worker,
             new SimpleLoop(0),
             $logger,
-            new PushMiddlewareDispatcher(new PushMiddlewareFactory($container, $callableFactory)),
+            new PushMiddlewareConfig(new PushMiddlewareFactory($container, $callableFactory)),
             $this->adapter,
         );
     }

tests/Integration/MiddlewareTest.php

@@ -23,8 +23,8 @@
 use Yiisoft\Queue\Middleware\FailureHandling\Implementation\ExponentialDelayMiddleware;
 use Yiisoft\Queue\Middleware\FailureHandling\Implementation\SendAgainMiddleware;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactory;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Tests\Integration\Support\TestMiddleware;
@@ -45,23 +45,25 @@ public function testFullStackPush(): void
             'channel 4',
         ];
 
-        $pushMiddlewareDispatcher = new PushMiddlewareDispatcher(
+        $pushMiddlewareConfig = new PushMiddlewareConfig(
             new PushMiddlewareFactory(
                 $this->createMock(ContainerInterface::class),
                 new CallableFactory(
                     $this->createMock(ContainerInterface::class),
                 ),
             ),
-            new TestMiddleware('common 1'),
-            new TestMiddleware('common 2'),
+            [
+                new TestMiddleware('common 1'),
+                new TestMiddleware('common 2'),
+            ],
         );
         $worker = $this->createMock(WorkerInterface::class);
         $worker->method('process')->willReturnArgument(0);
         $queue = new Queue(
             $worker,
             $this->createMock(LoopInterface::class),
             $this->createMock(LoggerInterface::class),
-            $pushMiddlewareDispatcher,
+            $pushMiddlewareConfig,
             name: 'test',
         );
         $queue = $queue

tests/TestCase.php

@@ -20,8 +20,8 @@
 use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareFactory;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactory;
+use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\Worker\Worker;
 use Yiisoft\Queue\Worker\WorkerInterface;
@@ -97,7 +97,7 @@ protected function createQueue(
             $this->getWorker(),
             $this->getLoop(),
             new NullLogger(),
-            $this->getPushMiddlewareDispatcher(),
+            $this->getPushMiddlewareConfig(),
             $adapter,
             $name,
         );
@@ -158,9 +158,9 @@ protected function getMessageHandlers(): array
         ];
     }
 
-    protected function getPushMiddlewareDispatcher(): PushMiddlewareDispatcher
+    protected function getPushMiddlewareConfig(): PushMiddlewareConfig
     {
-        return new PushMiddlewareDispatcher(
+        return new PushMiddlewareConfig(
             new PushMiddlewareFactory(
                 $this->getContainer(),
                 new CallableFactory($this->getContainer()),