diff --git a/CHANGELOG.md b/CHANGELOG.md index 58b0792e..9999835c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,4 +3,5 @@ ## 1.0.0 under development - Initial release. +- Bug #341: Pass the message a handler was registered for to that handler instead of the envelope an adapter wrapped it in, so a handler no longer reads properties off an `IdEnvelope`, `DelayEnvelope` or `FailureEnvelope` and works on empty values; what the envelope carried comes with the message as meta, and consume middleware still receive the envelope itself (@KalimeroMK) - Bug #347: `SignalLoop` installs signal handlers only while messages are consumed and restores the previous ones afterwards, instead of replacing them when it is created (@samdark) diff --git a/docs/guide/en/messages-and-handlers.md b/docs/guide/en/messages-and-handlers.md index 533e95dc..306ceb09 100644 --- a/docs/guide/en/messages-and-handlers.md +++ b/docs/guide/en/messages-and-handlers.md @@ -113,6 +113,12 @@ final class SendEmailHandler implements \Yiisoft\Queue\Message\Handler\HandlerIn The handler can have any dependencies injected through the DI container. The message payload remains plain data. +The handler is always given the message its type was registered for, never an envelope wrapped +around it on the way through the queue, so the `assert()` above holds on every adapter. What the +envelopes carried comes with it as message meta, so the handler still reads those values — +`IdEnvelope::fromMessage($message)->getId()`, or `$message->getMeta()` directly. Consume +middleware run before that step and receive the envelope itself. + ## Why the separation matters ### Versioning diff --git a/src/Middleware/Consume/ConsumeFinalHandler.php b/src/Middleware/Consume/ConsumeFinalHandler.php index 44c96e47..3da61943 100644 --- a/src/Middleware/Consume/ConsumeFinalHandler.php +++ b/src/Middleware/Consume/ConsumeFinalHandler.php @@ -5,6 +5,7 @@ namespace Yiisoft\Queue\Middleware\Consume; use Closure; +use Yiisoft\Queue\Message\Envelope; /** * @internal @@ -17,8 +18,19 @@ public function __construct( public function handleConsume(ConsumeRequest $request): ConsumeRequest { + $message = $request->getMessage(); + + // The handler is registered for a message type and is given that message, not whatever + // the adapter wrapped it in on the way. The envelope's meta is carried over rather than + // dropped with the wrapper: the envelope holds the merged meta while the message it wraps + // keeps its own, so unwrapping alone would lose the id, the delay and the failure counts. + // Middleware run before this step and still see the envelope itself. + if ($message instanceof Envelope) { + $message = $message->getMessage()->withMeta($message->getMeta()); + } + $handler = $this->handler; - $handler($request->getMessage()); + $handler($message); return $request; } diff --git a/tests/Unit/WorkerTest.php b/tests/Unit/WorkerTest.php index 0eea359b..f68a4480 100644 --- a/tests/Unit/WorkerTest.php +++ b/tests/Unit/WorkerTest.php @@ -12,11 +12,15 @@ use Yiisoft\Queue\Exception\MessageFailureException; use Yiisoft\Queue\Message\Handler\HandlerNotFoundException; use Yiisoft\Queue\Message\Handler\HandlerResolver; +use Yiisoft\Queue\Message\Envelope; use Yiisoft\Queue\Message\GenericMessage; +use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\Message\MessageInterface; +use Yiisoft\Queue\Middleware\Consume\ConsumeHandlerInterface; use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher; use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareFactoryInterface; use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareInterface; +use Yiisoft\Queue\Middleware\Consume\ConsumeRequest; use Yiisoft\Queue\Middleware\FailureHandling\FailureHandlingRequest; use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher; use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareFactoryInterface; @@ -140,6 +144,75 @@ public function testUnresolvableHandlerIsHandledByFailurePipeline(): void $worker->process($message, $queueName); } + public function testHandlerReceivesTheMessageAndNotTheEnvelopeAroundIt(): void + { + $message = new GenericMessage('simple', ['test-data']); + $envelope = new IdEnvelope($message, 'id-42'); + + $received = null; + $handlerResolver = $this->createHandlerResolver( + $message, + static function (MessageInterface $message) use (&$received): void { + $received = $message; + }, + ); + + $this->createWorkerByParams($handlerResolver)->process($envelope, 'test-queue'); + + // The handler is registered for the message type and gets that message, whatever the + // adapter wrapped it in on the way back. + $this->assertInstanceOf(GenericMessage::class, $received); + $this->assertNotInstanceOf(Envelope::class, $received); + $this->assertSame($message->getType(), $received->getType()); + $this->assertSame($message->getPayload(), $received->getPayload()); + + // What the envelope carried comes with it, so a handler that logs or de-duplicates on + // the id still has one. + $this->assertSame($envelope->getMeta(), $received->getMeta()); + $this->assertSame('id-42', IdEnvelope::fromMessage($received)->getId()); + } + + public function testConsumeMiddlewareStillSeesTheEnvelope(): void + { + $message = new GenericMessage('simple', null); + $envelope = new IdEnvelope($message, 'id-42'); + + $seenByMiddleware = null; + /** @var ConsumeMiddlewareInterface&MockObject $consumeMiddleware */ + $consumeMiddleware = $this->createMock(ConsumeMiddlewareInterface::class); + $consumeMiddleware + ->method('processConsume') + ->willReturnCallback( + static function (ConsumeRequest $request, ConsumeHandlerInterface $handler) use (&$seenByMiddleware) { + $seenByMiddleware = $request->getMessage(); + + return $handler->handleConsume($request); + }, + ); + + /** @var ConsumeMiddlewareFactoryInterface&MockObject $consumeMiddlewareFactory */ + $consumeMiddlewareFactory = $this->createMock(ConsumeMiddlewareFactoryInterface::class); + $consumeMiddlewareFactory->method('createConsumeMiddleware')->willReturn($consumeMiddleware); + $consumeDispatcher = new ConsumeMiddlewareDispatcher($consumeMiddlewareFactory, 'simple'); + + $receivedByHandler = null; + $handlerResolver = $this->createHandlerResolver( + $message, + static function (MessageInterface $message) use (&$receivedByHandler): void { + $receivedByHandler = $message; + }, + ); + + $worker = $this->createWorkerByParams($handlerResolver, new NullLogger(), $consumeDispatcher); + $worker->process($envelope, 'test-queue'); + + // Unwrapping happens at the last step, so middleware keep the envelope itself, and only + // the handler is given the message — with the meta the envelope carried. + $this->assertSame($envelope, $seenByMiddleware); + $this->assertNotInstanceOf(Envelope::class, $receivedByHandler); + $this->assertSame('id-42', IdEnvelope::fromMessage($receivedByHandler)->getId()); + } + private function createHandlerResolver(MessageInterface $message, callable $handler): HandlerResolver { $container = new SimpleContainer();