Revert some consume middlewares

src/Debug/QueueCollector.php

@@ -4,11 +4,11 @@
 
 namespace Yiisoft\Queue\Debug;
 
+use Yiisoft\Queue\Middleware\MiddlewareInterface;
 use Yiisoft\Yii\Debug\Collector\CollectorTrait;
 use Yiisoft\Yii\Debug\Collector\SummaryCollectorInterface;
 use Yiisoft\Queue\Enum\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Middleware\Push\MiddlewarePushInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueCollector implements SummaryCollectorInterface
@@ -53,7 +53,7 @@ public function collectStatus(string $id, JobStatus $status): void
     public function collectPush(
         string $channel,
         MessageInterface $message,
-        string|array|callable|MiddlewarePushInterface ...$middlewareDefinitions,
+        string|array|callable|MiddlewareInterface ...$middlewareDefinitions,
     ): void {
         if (!$this->isActive()) {
             return;

src/Worker/Worker.php

@@ -57,12 +57,12 @@ public function process(MessageInterface $message, QueueInterface $queue): Messa
         $request = new Request($message, $queue->getAdapter());
         $closure = fn (MessageInterface $message): mixed => $this->injector->invoke($handler, [$message]);
         try {
-            return $this->consumeMiddlewareDispatcher->dispatch($request, $this->createConsumeHandler($closure))->getMessage();
+            return $this->consumeMiddlewareDispatcher->dispatch($request, new ConsumeFinalHandler($closure))->getMessage();
         } catch (Throwable $exception) {
             $request = new FailureHandlingRequest($request->getMessage(), $exception, $queue);
 
             try {
-                $result = $this->failureMiddlewareDispatcher->dispatch($request, $this->createFailureHandler());
+                $result = $this->failureMiddlewareDispatcher->dispatch($request, new FailureFinalHandler());
                 $this->logger->info($exception->getMessage());
 
                 return $result->getMessage();
@@ -140,14 +140,4 @@ private function prepare(callable|object|array|string|null $definition): callabl
 
         return $definition;
     }
-
-    private function createConsumeHandler(Closure $handler): MessageHandlerInterface
-    {
-        return new ConsumeFinalHandler($handler);
-    }
-
-    private function createFailureHandler(): MessageFailureHandlerInterface
-    {
-        return new FailureFinalHandler();
-    }
 }

tests/Integration/Support/ConsumeMiddleware.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Tests\Integration\Support;
+
+use Yiisoft\Queue\Message\Message;
+use Yiisoft\Queue\Middleware\MessageHandlerInterface;
+use Yiisoft\Queue\Middleware\MiddlewareInterface;
+use Yiisoft\Queue\Middleware\Request;
+
+final class ConsumeMiddleware implements MiddlewareInterface
+{
+    public function __construct(private string $stage)
+    {
+    }
+
+    public function process(Request $request, MessageHandlerInterface $handler): Request
+    {
+        $message = $request->getMessage();
+        $stack = $message->getData();
+        $stack[] = $this->stage;
+        $messageNew = new Message($message->getHandlerName(), $stack);
+
+        return $handler->handle($request->withMessage($messageNew));
+    }
+}

tests/Unit/Middleware/Consume/ConsumeRequestTest.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Tests\Unit\Middleware\Consume;
+
+use Yiisoft\Queue\Adapter\AdapterInterface;
+use Yiisoft\Queue\Message\Message;
+use Yiisoft\Queue\Middleware\Request;
+use Yiisoft\Queue\QueueInterface;
+use Yiisoft\Queue\Tests\TestCase;
+
+final class ConsumeRequestTest extends TestCase
+{
+    public function testImmutable(): void
+    {
+        $message = new Message('test', 'test');
+        $adapter = $this->createMock(AdapterInterface::class);
+        $consumeRequest = new Request($message, $adapter);
+
+        $this->assertNotSame($consumeRequest, $consumeRequest->withMessage($message));
+        $this->assertNotSame($consumeRequest, $consumeRequest->withAdapter($adapter));
+    }
+}

tests/Unit/Middleware/Consume/InvalidMiddlewareDefinitionExceptionTest.php

@@ -0,0 +1,73 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Tests\Unit\Middleware\Consume;
+
+use PHPUnit\Framework\TestCase;
+use stdClass;
+use Yiisoft\Queue\Middleware\InvalidMiddlewareDefinitionException;
+use Yiisoft\Queue\Tests\Unit\Middleware\Support\TestCallableMiddleware;
+
+final class InvalidMiddlewareDefinitionExceptionTest extends TestCase
+{
+    public function dataBase(): array
+    {
+        return [
+            [
+                'test',
+                '"test"',
+            ],
+            [
+                new TestCallableMiddleware(),
+                sprintf(
+                    'an instance of "%s"',
+                    TestCallableMiddleware::class,
+                ),
+            ],
+            [
+                [TestCallableMiddleware::class, 'notExistsAction'],
+                sprintf(
+                    '["%s", "notExistsAction"]',
+                    TestCallableMiddleware::class,
+                ),
+            ],
+            [
+                ['class' => TestCallableMiddleware::class, 'index'],
+                sprintf(
+                    '["class" => "%s", "index"]',
+                    TestCallableMiddleware::class,
+                ),
+            ],
+        ];
+    }
+
+    /**
+     * @dataProvider dataBase
+     */
+    public function testBase(mixed $definition, string $expected): void
+    {
+        $exception = new InvalidMiddlewareDefinitionException($definition);
+        self::assertStringEndsWith('. Got ' . $expected . '.', $exception->getMessage());
+    }
+
+    public function dataUnknownDefinition(): array
+    {
+        return [
+            [42],
+            [[new stdClass()]],
+        ];
+    }
+
+    /**
+     * @dataProvider dataUnknownDefinition
+     */
+    public function testUnknownDefinition(mixed $definition): void
+    {
+        $exception = new InvalidMiddlewareDefinitionException($definition);
+        self::assertSame(
+            'Parameter should be either middleware class name or a callable.',
+            $exception->getMessage()
+        );
+    }
+}

tests/Unit/Middleware/Consume/MiddlewareDispatcherTest.php

@@ -0,0 +1,186 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Tests\Unit\Middleware\Consume;
+
+use PHPUnit\Framework\TestCase;
+use Psr\Container\ContainerInterface;
+use Yiisoft\Queue\Tests\Unit\Middleware\Support\TestCallableMiddleware;
+use Yiisoft\Queue\Tests\Unit\Middleware\Support\TestMiddleware;
+use Yiisoft\Test\Support\Container\SimpleContainer;
+use Yiisoft\Queue\Adapter\AdapterInterface;
+use Yiisoft\Queue\Message\Message;
+use Yiisoft\Queue\Middleware\CallableFactory;
+use Yiisoft\Queue\Middleware\MiddlewareDispatcher;
+use Yiisoft\Queue\Middleware\Request;
+use Yiisoft\Queue\Middleware\MessageHandlerInterface;
+use Yiisoft\Queue\Middleware\MiddlewareFactory;
+use Yiisoft\Queue\QueueInterface;
+use Yiisoft\Queue\Tests\App\FakeAdapter;
+
+final class MiddlewareDispatcherTest extends TestCase
+{
+    public function testCallableMiddlewareCalled(): void
+    {
+        $request = $this->getRequest();
+        $queue = $this->createMock(QueueInterface::class);
+        $adapter = $this->createMock(AdapterInterface::class);
+
+        $dispatcher = $this->createDispatcher()->withMiddlewares(
+            [
+                static function (Request $request) use ($adapter): Request {
+                    return $request->withMessage(new Message('test', 'New closure test data'))->withAdapter($adapter);
+                },
+            ]
+        );
+
+        $request = $dispatcher->dispatch($request, $this->getRequestHandler());
+        $this->assertSame('New closure test data', $request->getMessage()->getData());
+    }
+
+    public function testArrayMiddlewareCallableDefinition(): void
+    {
+        $request = $this->getRequest();
+        $container = $this->createContainer(
+            [
+                TestCallableMiddleware::class => new TestCallableMiddleware(),
+            ]
+        );
+        $dispatcher = $this->createDispatcher($container)->withMiddlewares([[TestCallableMiddleware::class, 'index']]);
+        $request = $dispatcher->dispatch($request, $this->getRequestHandler());
+        $this->assertSame('New test data', $request->getMessage()->getData());
+    }
+
+    public function testFactoryArrayDefinition(): void
+    {
+        $request = $this->getRequest();
+        $container = $this->createContainer();
+        $definition = [
+            'class' => TestMiddleware::class,
+            '__construct()' => ['message' => 'New test data from the definition'],
+        ];
+        $dispatcher = $this->createDispatcher($container)->withMiddlewares([$definition]);
+        $request = $dispatcher->dispatch($request, $this->getRequestHandler());
+        $this->assertSame('New test data from the definition', $request->getMessage()->getData());
+    }
+
+    public function testMiddlewareFullStackCalled(): void
+    {
+        $request = $this->getRequest();
+
+        $middleware1 = static function (Request $request, MessageHandlerInterface $handler): Request {
+            $request = $request->withMessage(new Message($request->getMessage()->getHandlerName(), 'new test data'));
+
+            return $handler->handle($request);
+        };
+        $middleware2 = static function (Request $request, MessageHandlerInterface $handler): Request {
+            $request = $request->withMessage(new Message('new handler', $request->getMessage()->getData()));
+
+            return $handler->handle($request);
+        };
+
+        $dispatcher = $this->createDispatcher()->withMiddlewares([$middleware1, $middleware2]);
+
+        $request = $dispatcher->dispatch($request, $this->getRequestHandler());
+        $this->assertSame('new test data', $request->getMessage()->getData());
+        $this->assertSame('new handler', $request->getMessage()->getHandlerName());
+    }
+
+    public function testMiddlewareStackInterrupted(): void
+    {
+        $request = $this->getRequest();
+
+        $middleware1 = static function (Request $request, MessageHandlerInterface $handler): Request {
+            return $request->withMessage(new Message($request->getMessage()->getHandlerName(), 'first'));
+        };
+        $middleware2 = static function (Request $request, MessageHandlerInterface $handler): Request {
+            return $request->withMessage(new Message($request->getMessage()->getHandlerName(), 'second'));
+        };
+
+        $dispatcher = $this->createDispatcher()->withMiddlewares([$middleware1, $middleware2]);
+
+        $request = $dispatcher->dispatch($request, $this->getRequestHandler());
+        $this->assertSame('first', $request->getMessage()->getData());
+    }
+
+    public function dataHasMiddlewares(): array
+    {
+        return [
+            [[], false],
+            [[[TestCallableMiddleware::class, 'index']], true],
+        ];
+    }
+
+    /**
+     * @dataProvider dataHasMiddlewares
+     */
+    public function testHasMiddlewares(array $definitions, bool $expected): void
+    {
+        self::assertSame(
+            $expected,
+            $this->createDispatcher()->withMiddlewares($definitions)->hasMiddlewares()
+        );
+    }
+
+    public function testImmutability(): void
+    {
+        $dispatcher = $this->createDispatcher();
+        self::assertNotSame($dispatcher, $dispatcher->withMiddlewares([]));
+    }
+
+    public function testResetStackOnWithMiddlewares(): void
+    {
+        $request = $this->getRequest();
+        $container = $this->createContainer(
+            [
+                TestCallableMiddleware::class => new TestCallableMiddleware(),
+                TestMiddleware::class => new TestMiddleware(),
+            ]
+        );
+
+        $dispatcher = $this
+            ->createDispatcher($container)
+            ->withMiddlewares([[TestCallableMiddleware::class, 'index']]);
+        $dispatcher->dispatch($request, $this->getRequestHandler());
+
+        $dispatcher = $dispatcher->withMiddlewares([TestMiddleware::class]);
+        $request = $dispatcher->dispatch($request, $this->getRequestHandler());
+
+        self::assertSame('New middleware test data', $request->getMessage()->getData());
+    }
+
+    private function getRequestHandler(): MessageHandlerInterface
+    {
+        return new class () implements MessageHandlerInterface {
+            public function handle(Request $request): Request
+            {
+                return $request;
+            }
+        };
+    }
+
+    private function createDispatcher(
+        ContainerInterface $container = null,
+    ): MiddlewareDispatcher {
+        $container ??= $this->createContainer([AdapterInterface::class => new FakeAdapter()]);
+        $callableFactory = new CallableFactory($container);
+
+        return new MiddlewareDispatcher(
+            new MiddlewareFactory($container, $callableFactory),
+        );
+    }
+
+    private function createContainer(array $instances = []): ContainerInterface
+    {
+        return new SimpleContainer($instances);
+    }
+
+    private function getRequest(): Request
+    {
+        return new Request(
+            new Message('handler', 'data'),
+            $this->createMock(AdapterInterface::class)
+        );
+    }
+}