diff --git a/pkg/pheanstalk/PheanstalkConsumer.php b/pkg/pheanstalk/PheanstalkConsumer.php index 4bcc7f4d2..93f24728a 100644 --- a/pkg/pheanstalk/PheanstalkConsumer.php +++ b/pkg/pheanstalk/PheanstalkConsumer.php @@ -47,8 +47,16 @@ public function getQueue() */ public function receive($timeout = 0) { - if ($job = $this->pheanstalk->reserveFromTube($this->destination->getName(), $timeout / 1000)) { - return $this->convertJobToMessage($job); + if ($timeout === 0) { + while (true) { + if ($job = $this->pheanstalk->reserveFromTube($this->destination->getName(), 5)) { + return $this->convertJobToMessage($job); + } + } + } else { + if ($job = $this->pheanstalk->reserveFromTube($this->destination->getName(), $timeout / 1000)) { + return $this->convertJobToMessage($job); + } } } @@ -101,7 +109,10 @@ public function reject(PsrMessage $message, $requeue = false) */ private function convertJobToMessage(Job $job) { + $stats = $this->pheanstalk->statsJob($job); + $message = PheanstalkMessage::jsonUnserialize($job->getData()); + $message->setRedelivered($stats['reserves'] > 1); $message->setJob($job); return $message;