123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174 |
- <?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\Bridge\Doctrine\Transport;
- use Doctrine\DBAL\DBALException;
- use Doctrine\DBAL\Exception;
- use Doctrine\DBAL\Exception\RetryableException;
- use Symfony\Component\Messenger\Envelope;
- use Symfony\Component\Messenger\Exception\LogicException;
- use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
- use Symfony\Component\Messenger\Exception\TransportException;
- use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
- use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
- use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
- use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
- use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
- use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
- /**
- * @author Vincent Touzet <vincent.touzet@gmail.com>
- */
- class DoctrineReceiver implements ReceiverInterface, MessageCountAwareInterface, ListableReceiverInterface
- {
- private const MAX_RETRIES = 3;
- private $retryingSafetyCounter = 0;
- private $connection;
- private $serializer;
- public function __construct(Connection $connection, SerializerInterface $serializer = null)
- {
- $this->connection = $connection;
- $this->serializer = $serializer ?? new PhpSerializer();
- }
- /**
- * {@inheritdoc}
- */
- public function get(): iterable
- {
- try {
- $doctrineEnvelope = $this->connection->get();
- $this->retryingSafetyCounter = 0; // reset counter
- } catch (RetryableException $exception) {
- // Do nothing when RetryableException occurs less than "MAX_RETRIES"
- // as it will likely be resolved on the next call to get()
- // Problem with concurrent consumers and database deadlocks
- if (++$this->retryingSafetyCounter >= self::MAX_RETRIES) {
- $this->retryingSafetyCounter = 0; // reset counter
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- return [];
- } catch (DBALException | Exception $exception) {
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- if (null === $doctrineEnvelope) {
- return [];
- }
- return [$this->createEnvelopeFromData($doctrineEnvelope)];
- }
- /**
- * {@inheritdoc}
- */
- public function ack(Envelope $envelope): void
- {
- try {
- $this->connection->ack($this->findDoctrineReceivedStamp($envelope)->getId());
- } catch (DBALException | Exception $exception) {
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- }
- /**
- * {@inheritdoc}
- */
- public function reject(Envelope $envelope): void
- {
- try {
- $this->connection->reject($this->findDoctrineReceivedStamp($envelope)->getId());
- } catch (DBALException | Exception $exception) {
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- }
- /**
- * {@inheritdoc}
- */
- public function getMessageCount(): int
- {
- try {
- return $this->connection->getMessageCount();
- } catch (DBALException | Exception $exception) {
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- }
- /**
- * {@inheritdoc}
- */
- public function all(int $limit = null): iterable
- {
- try {
- $doctrineEnvelopes = $this->connection->findAll($limit);
- } catch (DBALException | Exception $exception) {
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- foreach ($doctrineEnvelopes as $doctrineEnvelope) {
- yield $this->createEnvelopeFromData($doctrineEnvelope);
- }
- }
- /**
- * {@inheritdoc}
- */
- public function find($id): ?Envelope
- {
- try {
- $doctrineEnvelope = $this->connection->find($id);
- } catch (DBALException | Exception $exception) {
- throw new TransportException($exception->getMessage(), 0, $exception);
- }
- if (null === $doctrineEnvelope) {
- return null;
- }
- return $this->createEnvelopeFromData($doctrineEnvelope);
- }
- private function findDoctrineReceivedStamp(Envelope $envelope): DoctrineReceivedStamp
- {
- /** @var DoctrineReceivedStamp|null $doctrineReceivedStamp */
- $doctrineReceivedStamp = $envelope->last(DoctrineReceivedStamp::class);
- if (null === $doctrineReceivedStamp) {
- throw new LogicException('No DoctrineReceivedStamp found on the Envelope.');
- }
- return $doctrineReceivedStamp;
- }
- private function createEnvelopeFromData(array $data): Envelope
- {
- try {
- $envelope = $this->serializer->decode([
- 'body' => $data['body'],
- 'headers' => $data['headers'],
- ]);
- } catch (MessageDecodingFailedException $exception) {
- $this->connection->reject($data['id']);
- throw $exception;
- }
- return $envelope->with(
- new DoctrineReceivedStamp($data['id']),
- new TransportMessageIdStamp($data['id'])
- );
- }
- }
- class_alias(DoctrineReceiver::class, \Symfony\Component\Messenger\Transport\Doctrine\DoctrineReceiver::class);
|