From 0b337db8dd70230644d7ab7d65e94be9ca2236e8 Mon Sep 17 00:00:00 2001 From: Sergei Predvoditelev Date: Wed, 9 Sep 2026 12:19:59 +0300 Subject: [PATCH] Change `Worker::process()` return type to `void` No caller used the returned message, and its meaning was inconsistent: on the happy path it was the message after the consume middlewares, while on a handled failure it was the message re-queued by the failure pipeline. --- src/Debug/QueueWorkerInterfaceProxy.php | 7 +++++-- src/Worker/Worker.php | 8 +++----- src/Worker/WorkerInterface.php | 9 +++++++-- stubs/StubWorker.php | 4 +--- tests/Integration/MiddlewareTest.php | 13 ++++++++++--- .../Unit/Debug/QueueWorkerInterfaceProxyTest.php | 4 +--- tests/Unit/Stubs/StubWorkerTest.php | 12 +++--------- tests/Unit/WorkerTest.php | 16 +++++++++------- 8 files changed, 39 insertions(+), 34 deletions(-) diff --git a/src/Debug/QueueWorkerInterfaceProxy.php b/src/Debug/QueueWorkerInterfaceProxy.php index b26c5cd5..ed5a5f7a 100644 --- a/src/Debug/QueueWorkerInterfaceProxy.php +++ b/src/Debug/QueueWorkerInterfaceProxy.php @@ -8,6 +8,9 @@ use Yiisoft\Queue\QueueProducerInterface; use Yiisoft\Queue\Worker\WorkerInterface; +/** + * Debug proxy for {@see WorkerInterface} that records processed messages into the {@see QueueCollector}. + */ final class QueueWorkerInterfaceProxy implements WorkerInterface { public function __construct( @@ -19,8 +22,8 @@ public function process( MessageInterface $message, string $queueName, ?QueueProducerInterface $retryProducer = null, - ): MessageInterface { + ): void { $this->collector->collectWorkerProcessing($message, $queueName); - return $this->worker->process($message, $queueName, $retryProducer); + $this->worker->process($message, $queueName, $retryProducer); } } diff --git a/src/Worker/Worker.php b/src/Worker/Worker.php index b75eedfb..5b8eb8bc 100644 --- a/src/Worker/Worker.php +++ b/src/Worker/Worker.php @@ -34,7 +34,7 @@ public function process( MessageInterface $message, string $queueName, ?QueueProducerInterface $retryProducer = null, - ): MessageInterface { + ): void { $messageId = IdEnvelope::fromMessage($message)->getId(); if ($messageId === null) { $this->logger->info('Processing message without ID.'); @@ -46,15 +46,13 @@ public function process( try { $handler = $this->handlerResolver->resolve($message->getType()); $finishHandler = new ConsumeFinalHandler($handler->handle(...)); - return $this->consumeMiddlewareDispatcher->dispatch($request, $finishHandler)->getMessage(); + $this->consumeMiddlewareDispatcher->dispatch($request, $finishHandler); } catch (Throwable $exception) { $request = new FailureHandlingRequest($request->getMessage(), $exception, $request->getQueueName(), $retryProducer); try { - $result = $this->failureMiddlewareDispatcher->dispatch($request, new FailureFinalHandler()); + $this->failureMiddlewareDispatcher->dispatch($request, new FailureFinalHandler()); $this->logger->info($exception->getMessage()); - - return $result->getMessage(); } catch (Throwable $exception) { $exception = new MessageFailureException($message, $exception); $this->logger->error($exception->getMessage()); diff --git a/src/Worker/WorkerInterface.php b/src/Worker/WorkerInterface.php index 08ca5849..2e362f76 100644 --- a/src/Worker/WorkerInterface.php +++ b/src/Worker/WorkerInterface.php @@ -7,12 +7,17 @@ use Yiisoft\Queue\Message\MessageInterface; use Yiisoft\Queue\QueueProducerInterface; +/** + * Executes a message: runs it through the consume pipeline and, on failure, through the failure pipeline. + */ interface WorkerInterface { - /** @param string $queueName Logical execution queue name. */ + /** + * @param string $queueName Logical execution queue name. + */ public function process( MessageInterface $message, string $queueName, ?QueueProducerInterface $retryProducer = null, - ): MessageInterface; + ): void; } diff --git a/stubs/StubWorker.php b/stubs/StubWorker.php index 9e8bec68..85d8b831 100644 --- a/stubs/StubWorker.php +++ b/stubs/StubWorker.php @@ -17,7 +17,5 @@ public function process( MessageInterface $message, string $queueName, ?QueueProducerInterface $retryProducer = null, - ): MessageInterface { - return $message; - } + ): void {} } diff --git a/tests/Integration/MiddlewareTest.php b/tests/Integration/MiddlewareTest.php index 55e906bd..533ff875 100644 --- a/tests/Integration/MiddlewareTest.php +++ b/tests/Integration/MiddlewareTest.php @@ -95,17 +95,24 @@ public function testFullStackConsume(): void [], ); + $handledMessage = null; $worker = new Worker( new SimpleLogger(), $consumeMiddlewareDispatcher, $failureMiddlewareDispatcher, - new HandlerResolver(['test' => static fn() => true], $container), + new HandlerResolver( + ['test' => static function (MessageInterface $message) use (&$handledMessage): void { + $handledMessage = $message; + }], + $container, + ), ); $message = new GenericMessage('test', ['initial']); - $messageConsumed = $worker->process($message, 'test-queue'); + $worker->process($message, 'test-queue'); - self::assertEquals($stack, $messageConsumed->getPayload()); + self::assertInstanceOf(MessageInterface::class, $handledMessage); + self::assertEquals($stack, $handledMessage->getPayload()); } public function testFullStackFailure(): void diff --git a/tests/Unit/Debug/QueueWorkerInterfaceProxyTest.php b/tests/Unit/Debug/QueueWorkerInterfaceProxyTest.php index 51885015..1a6e83f6 100644 --- a/tests/Unit/Debug/QueueWorkerInterfaceProxyTest.php +++ b/tests/Unit/Debug/QueueWorkerInterfaceProxyTest.php @@ -19,9 +19,7 @@ public function testProcessDelegatesToWorker(): void $collector->startup(); $proxy = new QueueWorkerInterfaceProxy(new StubWorker(), $collector); - $result = $proxy->process($message, 'chan'); - - self::assertSame($message, $result); + $proxy->process($message, 'chan'); $collected = $collector->getCollected(); self::assertArrayHasKey('processingMessages', $collected); diff --git a/tests/Unit/Stubs/StubWorkerTest.php b/tests/Unit/Stubs/StubWorkerTest.php index e10b6cd1..6c7c059d 100644 --- a/tests/Unit/Stubs/StubWorkerTest.php +++ b/tests/Unit/Stubs/StubWorkerTest.php @@ -12,15 +12,9 @@ final class StubWorkerTest extends TestCase { public function testBase(): void { - $worker = new StubWorker(); - - $sourceMessage = new GenericMessage('test', 42); + $this->expectNotToPerformAssertions(); - $message = $worker->process($sourceMessage, 'test-queue'); - - $this->assertSame($sourceMessage, $message); - $this->assertSame('test', $message->getType()); - $this->assertSame(42, $message->getPayload()); - $this->assertSame([], $message->getMeta()); + $worker = new StubWorker(); + $worker->process(new GenericMessage('test', 42), 'test-queue'); } } diff --git a/tests/Unit/WorkerTest.php b/tests/Unit/WorkerTest.php index 8d5eae30..adab4e2f 100644 --- a/tests/Unit/WorkerTest.php +++ b/tests/Unit/WorkerTest.php @@ -94,7 +94,13 @@ public function testMessageFailureIsHandledSuccessfully(): void $finalMessage = new GenericMessage('final', null); /** @var FailureMiddlewareInterface&MockObject $failureMiddleware */ $failureMiddleware = $this->createMock(FailureMiddlewareInterface::class); - $failureMiddleware->method('processFailure')->willReturn(new FailureHandlingRequest($finalMessage, $originalException, $queueName)); + $failureMiddleware + ->expects(self::once()) + ->method('processFailure') + ->with(self::callback( + static fn(FailureHandlingRequest $request): bool => $request->getException() === $originalException, + )) + ->willReturn(new FailureHandlingRequest($finalMessage, $originalException, $queueName)); /** @var FailureMiddlewareFactoryInterface&MockObject $failureMiddlewareFactory */ $failureMiddlewareFactory = $this->createMock(FailureMiddlewareFactoryInterface::class); @@ -104,9 +110,7 @@ public function testMessageFailureIsHandledSuccessfully(): void $handlerResolver = $this->createHandlerResolver($message, static fn() => null); $worker = $this->createWorkerByParams($handlerResolver, new NullLogger(), $consumeDispatcher, $failureDispatcher); - $result = $worker->process($message, $queueName); - - self::assertSame($finalMessage, $result); + $worker->process($message, $queueName); } public function testUnresolvableHandlerIsHandledByFailurePipeline(): void @@ -133,9 +137,7 @@ public function testUnresolvableHandlerIsHandledByFailurePipeline(): void $worker = $this->createWorkerByParams($handlerResolver, failureMiddlewareDispatcher: $failureDispatcher); - $result = $worker->process($message, $queueName); - - self::assertSame($finalMessage, $result); + $worker->process($message, $queueName); } private function createHandlerResolver(MessageInterface $message, callable $handler): HandlerResolver