Introduce handler resolver

config/di.php

@@ -21,14 +21,14 @@
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactoryInterface;
+use Yiisoft\Queue\Message\Handler\HandlerResolver;
 use Yiisoft\Queue\Worker\Worker as QueueWorker;
 use Yiisoft\Queue\Worker\WorkerInterface;
 
 /* @var array $params */
 
 return [
-    QueueWorker::class => [
-        'class' => QueueWorker::class,
+    HandlerResolver::class => [
         '__construct()' => [$params['yiisoft/queue']['handlers']],
     ],
     WorkerInterface::class => QueueWorker::class,

config/params.php

@@ -9,7 +9,7 @@
 use Yiisoft\Queue\Debug\QueueConsumerProviderProxy;
 use Yiisoft\Queue\Debug\QueueProducerProviderProxy;
 use Yiisoft\Queue\Debug\QueueWorkerInterfaceProxy;
-use Yiisoft\Queue\Message\MessageHandlerInterface;
+use Yiisoft\Queue\Message\Handler\HandlerInterface;
 use Yiisoft\Queue\Message\Serializer\MessageSerializer;
 use Yiisoft\Queue\Provider\QueueConsumerProviderInterface;
 use Yiisoft\Queue\Provider\QueueProducerProviderInterface;
@@ -35,7 +35,7 @@
         'messages' => [],
         /**
          * Map of message type to handler. The worker uses this to find the handler for a received message.
-         * A handler may be a class name implementing {@see MessageHandlerInterface}, a callable, or any definition
+         * A handler may be a class name implementing {@see HandlerInterface}, a callable, or any definition
          * supported by yiisoft/injector. Example:
          * [
          *     'send-email' => SendEmailHandler::class,

src/Message/Handler/CallableHandler.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message\Handler;
+
+use Yiisoft\Queue\Message\MessageInterface;
+
+/**
+ * Handles a message by invoking the given callable.
+ *
+ * @internal
+ */
+final class CallableHandler implements HandlerInterface
+{
+    /**
+     * @param callable $handler Callable invoked to handle a message.
+     *
+     * @psalm-param callable(MessageInterface): void $handler
+     */
+    public function __construct(
+        private readonly mixed $handler,
+    ) {}
+
+    public function handle(MessageInterface $message): void
+    {
+        ($this->handler)($message);
+    }
+}

src/Message/Handler/HandlerInterface.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message\Handler;
+
+use Yiisoft\Queue\Message\MessageInterface;
+
+/**
+ * Handles a message.
+ */
+interface HandlerInterface
+{
+    /**
+     * Handle the given message.
+     */
+    public function handle(MessageInterface $message): void;
+}

src/Message/Handler/HandlerNotFoundException.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message\Handler;
+
+use LogicException;
+use Throwable;
+
+use function sprintf;
+
+/**
+ * Thrown when a handler for the given message type is not found.
+ */
+final class HandlerNotFoundException extends LogicException
+{
+    public function __construct(string $messageType, int $code = 0, ?Throwable $previous = null)
+    {
+        parent::__construct(
+            sprintf('Queue handler for message type "%s" does not exist.', $messageType),
+            $code,
+            $previous,
+        );
+    }
+}

src/Message/Handler/HandlerResolver.php

@@ -0,0 +1,144 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message\Handler;
+
+use LogicException;
+use Psr\Container\ContainerExceptionInterface;
+use Psr\Container\ContainerInterface;
+use Yiisoft\Injector\Injector;
+use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Middleware\CallableFactory;
+use Yiisoft\Queue\Middleware\InvalidCallableConfigurationException;
+
+use function array_key_exists;
+use function is_callable;
+use function is_string;
+use function sprintf;
+
+/**
+ * Resolves message handlers from configuration, a DI container, or a callable factory.
+ */
+final class HandlerResolver
+{
+    /**
+     * @var HandlerInterface[] Cache of resolved handlers.
+     * @psalm-var array<non-empty-string, HandlerInterface>
+     */
+    private array $cache = [];
+
+    private readonly Injector $injector;
+    private readonly CallableFactory $callableFactory;
+
+    /**
+     * @param (array|callable|HandlerInterface|string)[] $handlers Handler definitions indexed by message type.
+     * @param ContainerInterface $container Container used to resolve handlers.
+     * @param ContainerInterface|null $callableDependencyContainer Container used to resolve callable handler
+     * dependencies. If not set, the main container is used.
+     *
+     * @psalm-param array<non-empty-string, array|callable|HandlerInterface|string> $handlers
+     */
+    public function __construct(
+        private readonly array $handlers,
+        private readonly ContainerInterface $container,
+        ?ContainerInterface $callableDependencyContainer = null,
+    ) {
+        $this->injector = new Injector($callableDependencyContainer ?? $this->container);
+        $this->callableFactory = new CallableFactory($this->container);
+    }
+
+    /**
+     * Get a handler for the given message type.
+     *
+     * @param string $messageType Message type.
+     *
+     * @throws HandlerNotFoundException If no handler exists for the message type.
+     * @throws InvalidHandlerConfigurationException If the handler definition is configured incorrectly.
+     * @throws ContainerExceptionInterface Error while retrieving the entry from container.
+     */
+    public function resolve(string $messageType): HandlerInterface
+    {
+        if ($messageType === '') {
+            throw new LogicException('Message type cannot be empty.');
+        }
+
+        if (array_key_exists($messageType, $this->cache)) {
+            return $this->cache[$messageType];
+        }
+
+        $this->cache[$messageType] = $this->internalResolve($messageType);
+
+        return $this->cache[$messageType];
+    }
+
+    /**
+     * @throws HandlerNotFoundException
+     * @throws InvalidHandlerConfigurationException
+     * @throws ContainerExceptionInterface
+     */
+    private function internalResolve(string $messageType): HandlerInterface
+    {
+        $definition = $this->handlers[$messageType] ?? $messageType;
+
+        if ($definition instanceof HandlerInterface) {
+            return $definition;
+        }
+
+        if (is_string($definition)) {
+            return $this->getHandlerFromContainer($messageType, $definition);
+        }
+
+        return $this->createCallableHandler($messageType, $definition);
+    }
+
+    /**
+     * @throws HandlerNotFoundException
+     * @throws InvalidHandlerConfigurationException
+     * @throws ContainerExceptionInterface
+     */
+    private function getHandlerFromContainer(string $messageType, string $id): HandlerInterface
+    {
+        if (!$this->container->has($id)) {
+            throw new HandlerNotFoundException($messageType);
+        }
+
+        $handler = $this->container->get($id);
+
+        if ($handler instanceof HandlerInterface) {
+            return $handler;
+        }
+
+        if (is_callable($handler)) {
+            return $this->createCallableHandler($messageType, $handler);
+        }
+
+        throw new InvalidHandlerConfigurationException(
+            $messageType,
+            sprintf(
+                'Resolved from container handler should be an instance of "%s" or callable, got "%s".',
+                HandlerInterface::class,
+                get_debug_type($handler),
+            ),
+        );
+    }
+
+    /**
+     * @throws InvalidHandlerConfigurationException
+     * @throws ContainerExceptionInterface
+     */
+    private function createCallableHandler(string $messageType, mixed $definition): CallableHandler
+    {
+        try {
+            $callable = $this->callableFactory->create($definition);
+        } catch (InvalidCallableConfigurationException $exception) {
+            throw new InvalidHandlerConfigurationException($messageType, $exception->getMessage(), $exception);
+        }
+
+        $callable = function (MessageInterface $message) use ($callable): void {
+            $this->injector->invoke($callable, [$message]);
+        };
+
+        return new CallableHandler($callable);
+    }
+}

src/Message/Handler/InvalidHandlerConfigurationException.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message\Handler;
+
+use LogicException;
+use Throwable;
+
+use function sprintf;
+
+/**
+ * Thrown when a handler for the given message type is configured incorrectly.
+ */
+final class InvalidHandlerConfigurationException extends LogicException
+{
+    public function __construct(string $messageType, ?string $additionalMessage = null, ?Throwable $previous = null)
+    {
+        $message = sprintf('Queue handler for message type "%s" is configured incorrectly.', $messageType);
+
+        if ($additionalMessage !== null) {
+            $message .= ' ' . $additionalMessage;
+        }
+
+        parent::__construct($message, 0, $previous);
+    }
+}

src/Middleware/CallableFactory.php

@@ -4,10 +4,8 @@
 
 namespace Yiisoft\Queue\Middleware;
 
-use Closure;
 use Psr\Container\ContainerExceptionInterface;
 use Psr\Container\ContainerInterface;
-use Psr\Container\NotFoundExceptionInterface;
 use ReflectionException;
 use ReflectionMethod;
 
@@ -39,7 +37,7 @@ public function create(mixed $definition): callable
             throw new InvalidCallableConfigurationException();
         }
 
-        if ($definition instanceof Closure) {
+        if (is_callable($definition)) {
             return $definition;
         }
 
@@ -83,7 +81,6 @@ public function create(mixed $definition): callable
 
     /**
      * @throws ContainerExceptionInterface Error while retrieving the entry from container.
-     * @throws NotFoundExceptionInterface
      */
     private function fromDefinition(string $className, string $methodName): ?callable
     {

src/Middleware/Consume/ConsumeMiddlewareStack.php

@@ -65,9 +65,7 @@ public function __construct(
 
             public function handleConsume(ConsumeRequest $request): ConsumeRequest
             {
-                if ($this->middleware === null) {
-                    $this->middleware = ($this->middlewareFactory)();
-                }
+                $this->middleware ??= ($this->middlewareFactory)();
 
                 return $this->middleware->processConsume($request, $this->handler);
             }

src/Middleware/FailureHandling/FailureMiddlewareDispatcher.php

@@ -43,9 +43,7 @@ public function dispatch(
         }
         $definitions = array_reverse($this->middlewareDefinitions[$queueName]);
 
-        if (!isset($this->stack[$queueName])) {
-            $this->stack[$queueName] = new FailureMiddlewareStack($this->buildMiddlewares(...$definitions), $finishHandler);
-        }
+        $this->stack[$queueName] ??= new FailureMiddlewareStack($this->buildMiddlewares(...$definitions), $finishHandler);
 
         return $this->stack[$queueName]->handleFailure($request);
     }
@@ -83,9 +81,7 @@ public function withMiddlewares(array $middlewareDefinitions): self
 
     private function init(): void
     {
-        if (!isset($this->middlewareDefinitions[self::DEFAULT_PIPELINE])) {
-            $this->middlewareDefinitions[self::DEFAULT_PIPELINE] = [];
-        }
+        $this->middlewareDefinitions[self::DEFAULT_PIPELINE] ??= [];
     }
 
     /**

src/Middleware/FailureHandling/FailureMiddlewareStack.php

@@ -64,9 +64,7 @@ public function __construct(
 
             public function handleFailure(FailureHandlingRequest $request): FailureHandlingRequest
             {
-                if ($this->middleware === null) {
-                    $this->middleware = ($this->middlewareFactory)();
-                }
+                $this->middleware ??= ($this->middlewareFactory)();
 
                 return $this->middleware->processFailure($request, $this->handler);
             }

src/Middleware/Push/PushMiddlewareDispatcher.php

@@ -37,9 +37,7 @@ public function __construct(
      */
     public function dispatch(MessageInterface $message): MessageInterface
     {
-        if ($this->stack === null) {
-            $this->stack = new PushMiddlewareStack($this->buildMiddlewares(), $this->finishHandler);
-        }
+        $this->stack ??= new PushMiddlewareStack($this->buildMiddlewares(), $this->finishHandler);
 
         return $this->stack->handlePush($message);
     }

src/Middleware/Push/PushMiddlewareStack.php

@@ -68,9 +68,7 @@ public function __construct(
 
             public function handlePush(MessageInterface $message): MessageInterface
             {
-                if ($this->middleware === null) {
-                    $this->middleware = ($this->middlewareFactory)();
-                }
+                $this->middleware ??= ($this->middlewareFactory)();
 
                 return $this->middleware->processPush($message, $this->handler);
             }