nsq/nsq-bundle
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix: Prevent message duplication

Browse Source
This commit is contained in:
Konstantin Grachev
2021-07-08 18:34:17 +03:00
parent 4fac0e066f
commit 9d2ef8e132
1 changed files with 6 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/Messenger/NsqSender.php
View File

@ -39,25 +39,19 @@ final class NsqSender implements SenderInterface
$delayStamp = $envelope->last(DelayStamp::class); $delayStamp = $envelope->last(DelayStamp::class);
$delay = null !== $delayStamp ? $delayStamp->getDelay() : 0; $delay = null !== $delayStamp ? $delayStamp->getDelay() : 0;
$promise = null; if (null === $envelope->last(NsqReceivedStamp::class)) {
$encodedMessage = $this->serializer->encode($envelope);
$encodedMessage = json_encode($encodedMessage, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
if (null !== $envelope->last(NsqReceivedStamp::class)) { wait($producer->publish($this->topic, $encodedMessage, $delay));
} else {
$message = NsqReceivedStamp::getMessageFromEnvelope($envelope); $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
if (!$message->isProcessed()) { if (!$message->isProcessed()) {
$promise = $message->requeue($delay); wait($message->requeue($delay));
} }
} }
if (null === $promise) {
$encodedMessage = $this->serializer->encode($envelope->withoutAll(NsqReceivedStamp::class));
$encodedMessage = json_encode($encodedMessage, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
$promise = $producer->publish($this->topic, $encodedMessage, $delay);
}
wait($promise);
return $envelope; return $envelope;
} }