vendor/symfony/doctrine-messenger/Transport/DoctrineReceiver.php line 45

  1. <?php
  3. /*
  4.  * This file is part of the Symfony package.
  5.  *
  6.  * (c) Fabien Potencier <fabien@symfony.com>
  7.  *
  8.  * For the full copyright and license information, please view the LICENSE
  9.  * file that was distributed with this source code.
  10.  */
  12. namespace Symfony\Component\Messenger\Bridge\Doctrine\Transport;
  14. use Doctrine\DBAL\Exception as DBALException;
  15. use Doctrine\DBAL\Exception\RetryableException;
  16. use Symfony\Component\Messenger\Envelope;
  17. use Symfony\Component\Messenger\Exception\LogicException;
  18. use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
  19. use Symfony\Component\Messenger\Exception\TransportException;
  20. use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
  21. use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
  22. use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
  23. use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
  24. use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
  26. /**
  27.  * @author Vincent Touzet <vincent.touzet@gmail.com>
  28.  */
  29. class DoctrineReceiver implements ListableReceiverInterface, MessageCountAwareInterface
  30. {
  31.     private const MAX_RETRIES = 3;
  32.     private int $retryingSafetyCounter = 0;
  33.     private Connection $connection;
  34.     private SerializerInterface $serializer;
  36.     public function __construct(Connection $connection, SerializerInterface $serializer = null)
  37.     {
  38.         $this->connection = $connection;
  39.         $this->serializer = $serializer ?? new PhpSerializer();
  40.     }
  42.     public function get(): iterable
  43.     {
  44.         try {
  45.             $doctrineEnvelope = $this->connection->get();
  46.             $this->retryingSafetyCounter = 0; // reset counter
  47.         } catch (RetryableException $exception) {
  48.             // Do nothing when RetryableException occurs less than "MAX_RETRIES"
  49.             // as it will likely be resolved on the next call to get()
  50.             // Problem with concurrent consumers and database deadlocks
  51.             if (++$this->retryingSafetyCounter >= self::MAX_RETRIES) {
  52.                 $this->retryingSafetyCounter = 0; // reset counter
  53.                 throw new TransportException($exception->getMessage(), 0, $exception);
  54.             }
  56.             return [];
  57.         } catch (DBALException $exception) {
  58.             throw new TransportException($exception->getMessage(), 0, $exception);
  59.         }
  61.         if (null === $doctrineEnvelope) {
  62.             return [];
  63.         }
  65.         return [$this->createEnvelopeFromData($doctrineEnvelope)];
  66.     }
  68.     public function ack(Envelope $envelope): void
  69.     {
  70.         try {
  71.             $this->connection->ack($this->findDoctrineReceivedStamp($envelope)->getId());
  72.         } catch (DBALException $exception) {
  73.             throw new TransportException($exception->getMessage(), 0, $exception);
  74.         }
  75.     }
  77.     public function reject(Envelope $envelope): void
  78.     {
  79.         try {
  80.             $this->connection->reject($this->findDoctrineReceivedStamp($envelope)->getId());
  81.         } catch (DBALException $exception) {
  82.             throw new TransportException($exception->getMessage(), 0, $exception);
  83.         }
  84.     }
  86.     public function getMessageCount(): int
  87.     {
  88.         try {
  89.             return $this->connection->getMessageCount();
  90.         } catch (DBALException $exception) {
  91.             throw new TransportException($exception->getMessage(), 0, $exception);
  92.         }
  93.     }
  95.     public function all(int $limit = null): iterable
  96.     {
  97.         try {
  98.             $doctrineEnvelopes = $this->connection->findAll($limit);
  99.         } catch (DBALException $exception) {
  100.             throw new TransportException($exception->getMessage(), 0, $exception);
  101.         }
  103.         foreach ($doctrineEnvelopes as $doctrineEnvelope) {
  104.             yield $this->createEnvelopeFromData($doctrineEnvelope);
  105.         }
  106.     }
  108.     public function find(mixed $id): ?Envelope
  109.     {
  110.         try {
  111.             $doctrineEnvelope = $this->connection->find($id);
  112.         } catch (DBALException $exception) {
  113.             throw new TransportException($exception->getMessage(), 0, $exception);
  114.         }
  116.         if (null === $doctrineEnvelope) {
  117.             return null;
  118.         }
  120.         return $this->createEnvelopeFromData($doctrineEnvelope);
  121.     }
  123.     private function findDoctrineReceivedStamp(Envelope $envelope): DoctrineReceivedStamp
  124.     {
  125.         /** @var DoctrineReceivedStamp|null $doctrineReceivedStamp */
  126.         $doctrineReceivedStamp = $envelope->last(DoctrineReceivedStamp::class);
  128.         if (null === $doctrineReceivedStamp) {
  129.             throw new LogicException('No DoctrineReceivedStamp found on the Envelope.');
  130.         }
  132.         return $doctrineReceivedStamp;
  133.     }
  135.     private function createEnvelopeFromData(array $data): Envelope
  136.     {
  137.         try {
  138.             $envelope = $this->serializer->decode([
  139.                 'body' => $data['body'],
  140.                 'headers' => $data['headers'],
  141.             ]);
  142.         } catch (MessageDecodingFailedException $exception) {
  143.             $this->connection->reject($data['id']);
  145.             throw $exception;
  146.         }
  148.         return $envelope->with(
  149.             new DoctrineReceivedStamp($data['id']),
  150.             new TransportMessageIdStamp($data['id'])
  151.         );
  152.     }
  153. }