vendor/symfony/doctrine-messenger/Transport/DoctrineTransport.php line 72

  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\Connection as DbalConnection;
  15. use Doctrine\DBAL\Schema\Schema;
  16. use Doctrine\DBAL\Schema\Table;
  17. use Symfony\Component\Messenger\Envelope;
  18. use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
  19. use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
  20. use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
  21. use Symfony\Component\Messenger\Transport\SetupableTransportInterface;
  22. use Symfony\Component\Messenger\Transport\TransportInterface;
  24. /**
  25.  * @author Vincent Touzet <vincent.touzet@gmail.com>
  26.  */
  27. class DoctrineTransport implements TransportInterface, SetupableTransportInterface, MessageCountAwareInterface, ListableReceiverInterface
  28. {
  29.     private Connection $connection;
  30.     private SerializerInterface $serializer;
  31.     private DoctrineReceiver $receiver;
  32.     private DoctrineSender $sender;
  34.     public function __construct(Connection $connection, SerializerInterface $serializer)
  35.     {
  36.         $this->connection = $connection;
  37.         $this->serializer = $serializer;
  38.     }
  40.     public function get(): iterable
  41.     {
  42.         return $this->getReceiver()->get();
  43.     }
  45.     public function ack(Envelope $envelope): void
  46.     {
  47.         $this->getReceiver()->ack($envelope);
  48.     }
  50.     public function reject(Envelope $envelope): void
  51.     {
  52.         $this->getReceiver()->reject($envelope);
  53.     }
  55.     public function getMessageCount(): int
  56.     {
  57.         return $this->getReceiver()->getMessageCount();
  58.     }
  60.     public function all(int $limit = null): iterable
  61.     {
  62.         return $this->getReceiver()->all($limit);
  63.     }
  65.     public function find(mixed $id): ?Envelope
  66.     {
  67.         return $this->getReceiver()->find($id);
  68.     }
  70.     public function send(Envelope $envelope): Envelope
  71.     {
  72.         return $this->getSender()->send($envelope);
  73.     }
  75.     public function setup(): void
  76.     {
  77.         $this->connection->setup();
  78.     }
  80.     /**
  81.      * Adds the Table to the Schema if this transport uses this connection.
  82.      */
  83.     public function configureSchema(Schema $schema, DbalConnection $forConnection): void
  84.     {
  85.         $this->connection->configureSchema($schema, $forConnection);
  86.     }
  88.     /**
  89.      * Adds extra SQL if the given table was created by the Connection.
  90.      *
  91.      * @return string[]
  92.      */
  93.     public function getExtraSetupSqlForTable(Table $createdTable): array
  94.     {
  95.         return $this->connection->getExtraSetupSqlForTable($createdTable);
  96.     }
  98.     private function getReceiver(): DoctrineReceiver
  99.     {
  100.         return $this->receiver ??= new DoctrineReceiver($this->connection, $this->serializer);
  101.     }
  103.     private function getSender(): DoctrineSender
  104.     {
  105.         return $this->sender ??= new DoctrineSender($this->connection, $this->serializer);
  106.     }
  107. }