Replace queue role proxies with native debug middleware

README.md

@@ -37,7 +37,7 @@ For production use, you should install an adapter package that matches your mess
 See the [adapter list](docs/guide/en/adapter-list.md) and follow the adapter-specific documentation for installation and configuration details.
 
 > If you don't have an external broker — whether for development, testing, or because you want to
-> design around `QueueProducerInterface` from day one and add a real broker later — you can run the queue
+> start with a concrete producer and add a real broker later — you can run the queue
 > in [synchronous mode](docs/guide/en/synchronous-mode.md) using `SyncQueueProducer` instead of `AsyncQueueProducer`.
 > In this mode messages are processed immediately in the same process, so it won't provide true
 > async execution, but the code stays the same when you switch to a real adapter.
@@ -138,11 +138,11 @@ For setting up all classes manually, see the [Manual configuration](docs/guide/e
 To send a message to the queue, get the queue instance and call `push()`. Typically the queue is injected as a dependency:
 
 ```php
-use Yiisoft\Queue\QueueProducerInterface;
+use Yiisoft\Queue\AsyncQueueProducer;
 
 final readonly class Foo
 {
-    public function __construct(private QueueProducerInterface $queue) {}
+    public function __construct(private AsyncQueueProducer $queue) {}
 
     public function bar(): void
     {
@@ -166,10 +166,14 @@ By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/
 
 See [Console commands](docs/guide/en/console-commands.md) for more details.
 
-Producers use `Yiisoft\Queue\QueueProducerInterface` (`push()`, `status()`, `getQueueName()`); consumers use `Yiisoft\Queue\QueueConsumerInterface` (`run()`, `listen()`). See [capability configuration](docs/guide/en/queue-capabilities.md) for the strict role map used when named queues are configured.
+`AsyncQueueProducer` and `SyncQueueProducer` provide `push()`, `getStatus()`, and `getQueueName()`; `QueueConsumer` provides `run()` and `listen()`, while `Yiisoft\Queue\Worker\Worker` processes messages. For named queues, use the queue-keyed `QueueProducerStatusProvider` to obtain a producer's status capability. See [capability configuration](docs/guide/en/queue-capabilities.md) for the strict role map used when named queues are configured.
 
 > In case you're running the queue in synchronous mode (no adapter), `queue:listen` logs an info message and exits. The messages are processed immediately when pushed.
 
+#### Migration note
+
+`QueueProducerInterface`, `QueueConsumerInterface`, and `WorkerInterface` were removed. Custom implementations using these symbols are unsupported; use the concrete producer, consumer, and worker APIs instead.
+
 ## Documentation
 
 - [Guide](docs/guide/en/README.md)

composer.json

@@ -55,6 +55,7 @@
         "shipmonk/composer-dependency-analyser": "^1.8",
         "vimeo/psalm": "^5.26.1 || ^6.16.1",
         "yiisoft/code-style": "^1.0",
+        "yiisoft/di": "^1.4",
         "yiisoft/test-support": "^3.2.0",
         "yiisoft/yii-debug": "dev-master"
     },

config/di.php

@@ -21,17 +21,37 @@
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactoryInterface;
+use Yiisoft\Queue\Middleware\Worker\WorkerMiddlewareDispatcher;
+use Yiisoft\Queue\Middleware\Worker\WorkerMiddlewareFactory;
+use Yiisoft\Queue\Middleware\Worker\WorkerMiddlewareFactoryInterface;
+use Yiisoft\Queue\Provider\QueueProducerStatusProvider;
+use Yiisoft\Queue\Provider\QueueProducerStatusProviderInterface;
 use Yiisoft\Queue\Message\Handler\HandlerResolver;
-use Yiisoft\Queue\Worker\Worker as QueueWorker;
-use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Queue\Debug\Middleware\PushDebugMiddleware;
+use Yiisoft\Queue\Debug\Middleware\WorkerDebugMiddleware;
+use Yiisoft\Queue\Worker\Worker;
+use Yiisoft\Yii\Debug\Collector\SummaryCollectorInterface;
 
 /* @var array $params */
 
-return [
+$debugEnabled = (bool) ($params['yiisoft/yii-debug']['enabled'] ?? false)
+    && interface_exists(SummaryCollectorInterface::class);
+
+$pushMiddlewareDefinitions = array_merge(
+    $debugEnabled ? [PushDebugMiddleware::class] : [],
+    $params['yiisoft/queue']['middlewares-push'],
+);
+$workerMiddlewareDefinitions = array_merge(
+    $debugEnabled ? [WorkerDebugMiddleware::class] : [],
+    $params['yiisoft/queue']['middlewares-worker'] ?? [],
+);
+
+$definitions = [
     HandlerResolver::class => [
         '__construct()' => [$params['yiisoft/queue']['handlers']],
     ],
-    WorkerInterface::class => QueueWorker::class,
+    Worker::class => Worker::class,
+    QueueProducerStatusProviderInterface::class => QueueProducerStatusProvider::class,
     LoopInterface::class => static function (ContainerInterface $container): LoopInterface {
         return \extension_loaded('pcntl')
             ? $container->get(SignalLoop::class)
@@ -41,7 +61,11 @@
     ConsumeMiddlewareFactoryInterface::class => ConsumeMiddlewareFactory::class,
     FailureMiddlewareFactoryInterface::class => FailureMiddlewareFactory::class,
     PushMiddlewareConfig::class => [
-        '__construct()' => ['commonMiddlewareDefinitions' => $params['yiisoft/queue']['middlewares-push']],
+        '__construct()' => ['commonMiddlewareDefinitions' => $pushMiddlewareDefinitions],
+    ],
+    WorkerMiddlewareFactoryInterface::class => WorkerMiddlewareFactory::class,
+    WorkerMiddlewareDispatcher::class => [
+        '__construct()' => ['middlewareDefinitions' => $workerMiddlewareDefinitions],
     ],
     ConsumeMiddlewareDispatcher::class => [
         '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-consume']],
@@ -58,3 +82,5 @@
         ],
     ],
 ];
+
+return $definitions;

config/params.php

@@ -6,14 +6,10 @@
 use Yiisoft\Queue\Command\ListenCommand;
 use Yiisoft\Queue\Command\RunCommand;
 use Yiisoft\Queue\Debug\QueueCollector;
-use Yiisoft\Queue\Debug\QueueConsumerProviderProxy;
-use Yiisoft\Queue\Debug\QueueProducerProviderProxy;
-use Yiisoft\Queue\Debug\QueueWorkerInterfaceProxy;
+use Yiisoft\Queue\Debug\QueueProducerStatusProviderProxy;
 use Yiisoft\Queue\Message\Handler\HandlerInterface;
 use Yiisoft\Queue\Message\Serializer\MessageSerializer;
-use Yiisoft\Queue\Provider\QueueConsumerProviderInterface;
-use Yiisoft\Queue\Provider\QueueProducerProviderInterface;
-use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Queue\Provider\QueueProducerStatusProviderInterface;
 
 return [
     'yiisoft/yii-console' => [
@@ -44,6 +40,7 @@
          */
         'handlers' => [],
         'middlewares-push' => [],
+        'middlewares-worker' => [],
         'middlewares-consume' => [],
         'middlewares-fail' => [],
     ],
@@ -52,9 +49,7 @@
             QueueCollector::class,
         ],
         'trackedServices' => [
-            QueueProducerProviderInterface::class => [QueueProducerProviderProxy::class, QueueCollector::class],
-            QueueConsumerProviderInterface::class => [QueueConsumerProviderProxy::class, QueueCollector::class],
-            WorkerInterface::class => [QueueWorkerInterfaceProxy::class, QueueCollector::class],
+            QueueProducerStatusProviderInterface::class => [QueueProducerStatusProviderProxy::class, QueueCollector::class],
         ],
     ],
 ];

docs/guide/en/configuration-manual.md

@@ -94,25 +94,25 @@ $provider = new PredefinedQueueProvider([
 
 ## Running the queue
 
-Message consumption methods are available on `Yiisoft\Queue\QueueConsumerInterface`.
+Message consumption methods are available on `Yiisoft\Queue\QueueConsumer`.
 The producer and `QueueConsumer` are separate capabilities. Obtain or construct the consumer role before calling these methods.
 
 ### Processing existing messages
 
 ```php
-use Yiisoft\Queue\QueueConsumerInterface;
+use Yiisoft\Queue\QueueConsumer;
 
-/** @var QueueConsumerInterface $queue */
+/** @var QueueConsumer $queue */
 $queue->run();      // Process all messages
 $queue->run(10);    // Process up to 10 messages
 ```
 
 ### Listening for new messages
 
 ```php
-use Yiisoft\Queue\QueueConsumerInterface;
+use Yiisoft\Queue\QueueConsumer;
 
-/** @var QueueConsumerInterface $queue */
+/** @var QueueConsumer $queue */
 $queue->listen();   // Run indefinitely
 ```
 

docs/guide/en/debug-integration-advanced.md

@@ -1,49 +1,50 @@
 # Advanced Yii Debug integration
 
-Use this guide when you need to understand which events are tracked by the queue collector, how proxy services operate, and how to wire the collector manually.
+Use this guide to understand the optional native middleware and status-provider instrumentation used by the queue collector.
 
 ## What is collected
 
-The integration is based on `Yiisoft\Queue\Debug\QueueCollector` and captures:
+`Yiisoft\Queue\Debug\QueueCollector` can capture:
 
-- Pushed messages grouped by queue name.
-- Message status checks performed via `QueueProducerInterface::status()`.
-- Messages processed by a worker grouped by queue name.
+- pushed messages, grouped by their normalized queue key;
+- status checks made through the producer status capability;
+- worker processing events, grouped by queue key.
 
-## How it works
+There is no consumer processing metric or debug proxy. Push instrumentation does not report status checks, and status instrumentation does not wrap push calls.
 
-The collector is enabled by registering it in Yii Debug and wrapping tracked services with proxy implementations.
+## How it works
 
-Out of the box (see this package's `config/params.php`), the following services are wrapped:
+The integration uses native middleware:
 
-- `Yiisoft\Queue\Provider\QueueProducerProviderInterface` is wrapped with `Yiisoft\Queue\Debug\QueueProducerProviderProxy`, which returns `QueueProducerDecorator` instances so `push()` and `status()` calls are reported.
-- `Yiisoft\Queue\Provider\QueueConsumerProviderInterface` is wrapped with `Yiisoft\Queue\Debug\QueueConsumerProviderProxy`, which returns typed consumer decorators.
-- `Yiisoft\Queue\Worker\WorkerInterface` is wrapped with `Yiisoft\Queue\Debug\QueueWorkerInterfaceProxy` to record message processing events.
+- `Yiisoft\Queue\Debug\Middleware\PushDebugMiddleware` wraps the push pipeline. After the downstream/final push handler returns, it records the message from the returned `PushRequest` and the immutable normalized identity from the incoming `PushRequest`. The returned message is not necessarily adapter output: synchronous pushing may process and replace it. If downstream processing throws, no push event is recorded; a middleware that short-circuits before this middleware is reached also bypasses it.
+- `Yiisoft\Queue\Debug\Middleware\WorkerDebugMiddleware` runs in the separate worker pipeline and records processing before handler resolution. The processing event is recorded before downstream handling, so a later exception does not remove it; middleware that short-circuits before this middleware is reached prevents it.
+- `Yiisoft\Queue\Debug\QueueProducerStatusProviderProxy` wraps only `QueueProducerStatusProviderInterface`. It decorates the status capability returned by `getStatus($queueName)`; status-provider operations are not push or consumer-processing instrumentation.
 
-To see data in the debug panel, obtain the typed provider dependencies and `WorkerInterface` from the DI container — the proxies are registered there and will not be active if the services are instantiated directly.
+All of these are optional. Services instantiated directly are instrumented only when the corresponding middleware or provider wrapper is explicitly supplied.
 
 ## Manual configuration
 
-If you do not rely on the defaults supplied via [yiisoft/config](https://github.com/yiisoft/config), configure the collector and proxies explicitly:
+When using [yiisoft/config](https://github.com/yiisoft/config), configure `middlewares-push` and `middlewares-worker` as needed. Register the status provider wrapper only if status instrumentation is wanted:
 
 ```php
+use Yiisoft\Queue\Debug\Middleware\PushDebugMiddleware;
+use Yiisoft\Queue\Debug\Middleware\WorkerDebugMiddleware;
 use Yiisoft\Queue\Debug\QueueCollector;
-use Yiisoft\Queue\Debug\QueueConsumerProviderProxy;
-use Yiisoft\Queue\Debug\QueueProducerProviderProxy;
-use Yiisoft\Queue\Debug\QueueWorkerInterfaceProxy;
-use Yiisoft\Queue\Provider\QueueConsumerProviderInterface;
-use Yiisoft\Queue\Provider\QueueProducerProviderInterface;
-use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Queue\Debug\QueueProducerStatusProviderProxy;
+use Yiisoft\Queue\Provider\QueueProducerStatusProviderInterface;
 
 return [
+    'yiisoft/queue' => [
+        'middlewares-push' => [PushDebugMiddleware::class],
+        'middlewares-worker' => [WorkerDebugMiddleware::class],
+    ],
     'yiisoft/yii-debug' => [
-        'collectors' => [
-            QueueCollector::class,
-        ],
+        'collectors' => [QueueCollector::class],
         'trackedServices' => [
-            QueueProducerProviderInterface::class => [QueueProducerProviderProxy::class, QueueCollector::class],
-            QueueConsumerProviderInterface::class => [QueueConsumerProviderProxy::class, QueueCollector::class],
-            WorkerInterface::class => [QueueWorkerInterfaceProxy::class, QueueCollector::class],
+            QueueProducerStatusProviderInterface::class => [
+                QueueProducerStatusProviderProxy::class,
+                QueueCollector::class,
+            ],
         ],
     ],
 ];

docs/guide/en/debug-integration.md

@@ -1,11 +1,9 @@
 # Yii Debug integration
 
-This package provides an integration with [yiisoft/yii-debug](https://github.com/yiisoft/yii-debug).
+This package provides optional integration with [yiisoft/yii-debug](https://github.com/yiisoft/yii-debug).
 
-When Yii Debug is enabled, the queue collector adds a panel that shows pushed messages, message status checks, and worker activity.
+When enabled, optional native push and worker middleware add queue events to the queue collector. Push events contain the normalized queue key and the message from the final push handler's returned `PushRequest`; this is not necessarily a message returned by an adapter (for example, synchronous pushing can process and replace the message). Worker events are recorded before handler resolution. A push event is recorded only when the downstream push handler returns; an exception prevents that event, and middleware that short-circuits before the debug middleware is reached bypasses it. Status instrumentation is separate and wraps only the `QueueProducerStatusProviderInterface` capability.
 
-If you use [yiisoft/config](https://github.com/yiisoft/config) together with this package, the debug collector is registered automatically. For manual configuration snippets and proxy wiring details, see [Advanced Yii Debug integration](debug-integration-advanced.md).
+The instrumentation is optional: applications can enable or omit each middleware and the status provider wrapper independently. There are no producer, consumer, or worker debug proxy/decorator services.
 
-## See also
-
-- [Advanced Yii Debug integration](debug-integration-advanced.md) — collector internals, proxies, and manual wiring
+For manual wiring and tracked events, see [Advanced Yii Debug integration](debug-integration-advanced.md).

docs/guide/en/error-handling-advanced.md

@@ -19,7 +19,7 @@ This document covers advanced internals of the failure handling pipeline, built-
     - the message
     - the caught exception
     - the logical queue name
-    - an optional direct retry producer (provided for synchronous producer execution)
+    - an optional direct retry target, a `Closure(MessageInterface): MessageInterface` (provided by the worker when available)
 
 4. A failure pipeline is selected by queue name
 
@@ -84,7 +84,7 @@ This interface has the only method `processFailure` with these parameters:
 - [`FailureHandlingRequest $request`](../../../src/Middleware/FailureHandling/FailureHandlingRequest.php) - a request for a message handling. It consists of
     - a [message](../../../src/Message/MessageInterface.php)
     - a `Throwable $exception` object thrown on the `request` handling
-    - the logical queue name the message came from and, when available, a direct retry producer
+    - the logical queue name the message came from and, when available, a direct retry target closure (`Closure(MessageInterface): MessageInterface`)
 - `FailureHandlerInterface $handler` - failure strategy pipeline continuation. Your Middleware should call `$handler->handleFailure($request)` when the middleware itself should not interrupt failure pipeline execution.
 
-> Note: your strategy have to check by its own if it should be applied. Look into [`SendAgainMiddleware::suits()`](../../../src/Middleware/FailureHandling/Implementation/SendAgainMiddleware.php#L54) for an example.
+> Note: your strategy have to check by its own if it should be applied. Look into [`SendAgainMiddleware::suits()`](../../../src/Middleware/FailureHandling/Implementation/SendAgainMiddleware.php#L64) for an example.

docs/guide/en/error-handling.md

@@ -38,7 +38,9 @@ Here below is configuration via [yiisoft/config](https://github.com/yiisoft/conf
             static fn (QueueProducerProviderInterface $queues) => new SendAgainMiddleware(
                 id: 'default-second-resend',
                 maxAttempts: 1,
-                targetQueue: $queues->getProducer('failed-messages'),
+                targetQueue: static function (MessageInterface $message) use ($queues): MessageInterface {
+                    return $queues->getProducer('failed-messages')->push($message);
+                },
             ),
         ],
         'failed-messages' => [
@@ -87,8 +89,8 @@ Failures of messages that arrived in the `failed-messages` queue directly (bypas
 
  - `id` - A unique string. Allows to use this strategy more than once for the same message, just like in example above.
  - `maxAttempts` - Maximum attempts count for this strategy with the given $id before it will give up.
- - `targetQueue` - An optional `QueueProducerInterface` for an explicit retry destination. When it is `null`, synchronous execution supplies its originating producer; asynchronous execution resolves the originating queue name through `producerProvider`.
- - `producerProvider` - The `QueueProducerProviderInterface` used to resolve the source producer for asynchronous retries when no `targetQueue` is supplied. Configure it, or provide `targetQueue`; otherwise retry fails with a configuration error.
+ - `targetQueue` - An optional `Closure(MessageInterface): MessageInterface` retry target. The closure receives the failed message, pushes it to the target queue, and returns the message returned by the producer.
+ - `producerProvider` - The `QueueProducerProviderInterface` used to create a retry closure for the originating queue when no `targetQueue` or worker-supplied retry closure is available. Configure it, or provide `targetQueue`; otherwise retry fails with a configuration error.
 
  State tracking:
 
@@ -106,8 +108,8 @@ It's configured via constructor parameters, too. Here they are:
  - `delayInitial` - The initial delay that will be applied to a message for the first time. It must be a positive float. 
  - `delayMaximum` - The maximum delay which can be applied to a single message. Must be above the `delayInitial`.
  - `exponent` - Message handling delay will be multiplied by exponent each time it fails.
- - `queue` - An optional `QueueProducerInterface` retry destination. When it is `null`, synchronous execution supplies its originating producer; asynchronous execution resolves the originating queue name through `producerProvider`.
- - `producerProvider` - The `QueueProducerProviderInterface` used for that asynchronous source-producer lookup.
+ - `queue` - An optional `Closure(MessageInterface): MessageInterface` retry target. The closure receives the failed message, pushes it to the target queue, and returns the message returned by the producer.
+ - `producerProvider` - The `QueueProducerProviderInterface` used to create a retry closure for the originating queue when no `queue` or worker-supplied retry closure is available.
 
  Requirements: