Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
6 changes: 6 additions & 0 deletions docs/guide/en/messages-and-handlers.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
14 changes: 13 additions & 1 deletion src/Middleware/Consume/ConsumeFinalHandler.php
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
namespace Yiisoft\Queue\Middleware\Consume;

use Closure;
use Yiisoft\Queue\Message\Envelope;

/**
* @internal
Expand All @@ -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;
}
Expand Down
73 changes: 73 additions & 0 deletions tests/Unit/WorkerTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down
Loading