Non retry messages must be acknowledged
This commit is contained in:
37
src/Messenger/AckUnrecoverableMessageListener.php
Normal file
37
src/Messenger/AckUnrecoverableMessageListener.php
Normal file
@ -0,0 +1,37 @@
|
||||
<?php
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Nsq\NsqBundle\Messenger;
|
||||
|
||||
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
|
||||
use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
|
||||
use function Amp\Promise\wait;
|
||||
|
||||
final class AckUnrecoverableMessageListener implements EventSubscriberInterface
|
||||
{
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public static function getSubscribedEvents(): array
|
||||
{
|
||||
return [
|
||||
WorkerMessageFailedEvent::class => ['onMessageFailed', 500],
|
||||
];
|
||||
}
|
||||
|
||||
public function onMessageFailed(WorkerMessageFailedEvent $event): void
|
||||
{
|
||||
if ($event->willRetry()) {
|
||||
return;
|
||||
}
|
||||
|
||||
$envelope = $event->getEnvelope();
|
||||
$message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
|
||||
|
||||
if ($message->isProcessed()) {
|
||||
return;
|
||||
}
|
||||
|
||||
wait($message->finish());
|
||||
}
|
||||
}
|
@ -3,3 +3,7 @@ services:
|
||||
tags:
|
||||
- 'messenger.transport_factory'
|
||||
- 'container.no_preload'
|
||||
|
||||
Nsq\NsqBundle\Messenger\AckUnrecoverableMessageListener:
|
||||
tags:
|
||||
- { name: kernel.event_subscriber }
|
||||
|
Reference in New Issue
Block a user