2017-09-30 13:16:30 +01:00
|
|
|
<?php
|
|
|
|
|
|
|
|
/*
|
|
|
|
* This file is part of the Symfony package.
|
|
|
|
*
|
|
|
|
* (c) Fabien Potencier <fabien@symfony.com>
|
|
|
|
*
|
|
|
|
* For the full copyright and license information, please view the LICENSE
|
|
|
|
* file that was distributed with this source code.
|
|
|
|
*/
|
|
|
|
|
|
|
|
namespace Symfony\Component\Messenger;
|
|
|
|
|
2019-03-13 19:48:28 +00:00
|
|
|
use Psr\Log\LoggerInterface;
|
|
|
|
use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
|
|
|
|
use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
|
|
|
|
use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
|
|
|
|
use Symfony\Component\Messenger\Exception\LogicException;
|
|
|
|
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
|
|
|
|
use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
|
|
|
|
use Symfony\Component\Messenger\Stamp\DelayStamp;
|
2018-10-18 09:53:38 +01:00
|
|
|
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
|
2019-03-13 19:48:28 +00:00
|
|
|
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
|
|
|
|
use Symfony\Component\Messenger\Stamp\SentStamp;
|
2018-10-22 09:20:29 +01:00
|
|
|
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
|
2019-03-25 16:04:58 +00:00
|
|
|
use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
|
2017-09-30 13:16:30 +01:00
|
|
|
|
|
|
|
/**
|
|
|
|
* @author Samuel Roze <samuel.roze@gmail.com>
|
2018-10-31 05:40:28 +00:00
|
|
|
*
|
|
|
|
* @experimental in 4.2
|
2018-11-24 10:21:28 +00:00
|
|
|
*
|
|
|
|
* @final
|
2017-09-30 13:16:30 +01:00
|
|
|
*/
|
|
|
|
class Worker
|
|
|
|
{
|
|
|
|
private $receiver;
|
|
|
|
private $bus;
|
2019-03-13 19:48:28 +00:00
|
|
|
private $receiverName;
|
|
|
|
private $retryStrategy;
|
|
|
|
private $eventDispatcher;
|
|
|
|
private $logger;
|
2017-09-30 13:16:30 +01:00
|
|
|
|
2019-03-13 19:48:28 +00:00
|
|
|
public function __construct(ReceiverInterface $receiver, MessageBusInterface $bus, string $receiverName = null, RetryStrategyInterface $retryStrategy = null, EventDispatcherInterface $eventDispatcher = null, LoggerInterface $logger = null)
|
2017-09-30 13:16:30 +01:00
|
|
|
{
|
|
|
|
$this->receiver = $receiver;
|
|
|
|
$this->bus = $bus;
|
2019-03-13 19:48:28 +00:00
|
|
|
if (null === $receiverName) {
|
|
|
|
@trigger_error(sprintf('Instantiating the "%s" class without passing a third argument is deprecated since Symfony 4.3.', __CLASS__), E_USER_DEPRECATED);
|
|
|
|
|
|
|
|
$receiverName = 'unknown';
|
|
|
|
}
|
|
|
|
$this->receiverName = $receiverName;
|
|
|
|
$this->retryStrategy = $retryStrategy;
|
|
|
|
$this->eventDispatcher = $eventDispatcher;
|
|
|
|
$this->logger = $logger;
|
2017-09-30 13:16:30 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Receive the messages and dispatch them to the bus.
|
|
|
|
*/
|
|
|
|
public function run()
|
|
|
|
{
|
2018-04-12 16:12:27 +01:00
|
|
|
if (\function_exists('pcntl_signal')) {
|
2018-03-13 16:34:51 +00:00
|
|
|
pcntl_signal(SIGTERM, function () {
|
|
|
|
$this->receiver->stop();
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2018-05-07 16:30:59 +01:00
|
|
|
$this->receiver->receive(function (?Envelope $envelope) {
|
|
|
|
if (null === $envelope) {
|
2019-03-13 19:48:28 +00:00
|
|
|
if (\function_exists('pcntl_signal_dispatch')) {
|
|
|
|
pcntl_signal_dispatch();
|
|
|
|
}
|
|
|
|
|
2018-03-13 16:34:51 +00:00
|
|
|
return;
|
|
|
|
}
|
2017-09-30 13:16:30 +01:00
|
|
|
|
2019-03-13 19:48:28 +00:00
|
|
|
$this->dispatchEvent(new WorkerMessageReceivedEvent($envelope, $this->receiverName));
|
|
|
|
|
|
|
|
$message = $envelope->getMessage();
|
|
|
|
$context = [
|
|
|
|
'message' => $message,
|
|
|
|
'class' => \get_class($message),
|
|
|
|
];
|
|
|
|
|
|
|
|
try {
|
|
|
|
$envelope = $this->bus->dispatch($envelope->with(new ReceivedStamp()));
|
|
|
|
} catch (\Throwable $throwable) {
|
|
|
|
$shouldRetry = $this->shouldRetry($throwable, $envelope);
|
|
|
|
|
|
|
|
$this->dispatchEvent(new WorkerMessageFailedEvent($envelope, $this->receiverName, $throwable, $shouldRetry));
|
|
|
|
|
|
|
|
if ($shouldRetry) {
|
|
|
|
if (null === $this->retryStrategy) {
|
|
|
|
// not logically allowed, but check just in case
|
|
|
|
throw new LogicException('Retrying is not supported without a retry strategy.');
|
|
|
|
}
|
|
|
|
|
|
|
|
$retryCount = $this->getRetryCount($envelope) + 1;
|
|
|
|
if (null !== $this->logger) {
|
|
|
|
$this->logger->info('Retrying {class} - retry #{retryCount}.', $context + ['retryCount' => $retryCount, 'error' => $throwable]);
|
|
|
|
}
|
|
|
|
|
|
|
|
// add the delay and retry stamp info + remove ReceivedStamp
|
|
|
|
$retryEnvelope = $envelope->with(new DelayStamp($this->retryStrategy->getWaitingTime($envelope)))
|
|
|
|
->with(new RedeliveryStamp($retryCount, $this->getSenderAlias($envelope)))
|
|
|
|
->withoutAll(ReceivedStamp::class);
|
|
|
|
|
|
|
|
// re-send the message
|
|
|
|
$this->bus->dispatch($retryEnvelope);
|
|
|
|
// acknowledge the previous message has received
|
|
|
|
$this->receiver->ack($envelope);
|
|
|
|
} else {
|
|
|
|
if (null !== $this->logger) {
|
|
|
|
$this->logger->info('Rejecting {class} (removing from transport).', $context + ['error' => $throwable]);
|
|
|
|
}
|
|
|
|
|
|
|
|
$this->receiver->reject($envelope);
|
|
|
|
}
|
|
|
|
|
|
|
|
if (\function_exists('pcntl_signal_dispatch')) {
|
|
|
|
pcntl_signal_dispatch();
|
|
|
|
}
|
|
|
|
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
$this->dispatchEvent(new WorkerMessageHandledEvent($envelope, $this->receiverName));
|
|
|
|
|
|
|
|
if (null !== $this->logger) {
|
|
|
|
$this->logger->info('{class} was handled successfully (acknowledging to transport).', $context);
|
|
|
|
}
|
|
|
|
|
|
|
|
$this->receiver->ack($envelope);
|
|
|
|
|
|
|
|
if (\function_exists('pcntl_signal_dispatch')) {
|
|
|
|
pcntl_signal_dispatch();
|
|
|
|
}
|
2018-03-13 16:34:51 +00:00
|
|
|
});
|
2017-09-30 13:16:30 +01:00
|
|
|
}
|
2019-03-13 19:48:28 +00:00
|
|
|
|
2019-03-25 16:04:58 +00:00
|
|
|
private function dispatchEvent($event)
|
2019-03-13 19:48:28 +00:00
|
|
|
{
|
|
|
|
if (null === $this->eventDispatcher) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
$this->eventDispatcher->dispatch($event);
|
|
|
|
}
|
|
|
|
|
|
|
|
private function shouldRetry(\Throwable $e, Envelope $envelope): bool
|
|
|
|
{
|
|
|
|
if ($e instanceof UnrecoverableMessageHandlingException) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
|
|
|
|
if (null === $this->retryStrategy) {
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
|
|
|
|
return $this->retryStrategy->isRetryable($envelope);
|
|
|
|
}
|
|
|
|
|
|
|
|
private function getRetryCount(Envelope $envelope): int
|
|
|
|
{
|
|
|
|
/** @var RedeliveryStamp|null $retryMessageStamp */
|
|
|
|
$retryMessageStamp = $envelope->last(RedeliveryStamp::class);
|
|
|
|
|
|
|
|
return $retryMessageStamp ? $retryMessageStamp->getRetryCount() : 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
private function getSenderAlias(Envelope $envelope): ?string
|
|
|
|
{
|
|
|
|
/** @var SentStamp|null $sentStamp */
|
|
|
|
$sentStamp = $envelope->last(SentStamp::class);
|
|
|
|
|
|
|
|
return $sentStamp ? $sentStamp->getSenderAlias() : null;
|
|
|
|
}
|
2017-09-30 13:16:30 +01:00
|
|
|
}
|