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
  {