DoctrineTransport.php 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. <?php
  2. /*
  3. * This file is part of the Symfony package.
  4. *
  5. * (c) Fabien Potencier <fabien@symfony.com>
  6. *
  7. * For the full copyright and license information, please view the LICENSE
  8. * file that was distributed with this source code.
  9. */
  10. namespace Symfony\Component\Messenger\Bridge\Doctrine\Transport;
  11. use Doctrine\DBAL\Connection as DbalConnection;
  12. use Doctrine\DBAL\Schema\Schema;
  13. use Doctrine\DBAL\Schema\Table;
  14. use Symfony\Component\Messenger\Envelope;
  15. use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
  16. use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
  17. use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
  18. use Symfony\Component\Messenger\Transport\SetupableTransportInterface;
  19. use Symfony\Component\Messenger\Transport\TransportInterface;
  20. /**
  21. * @author Vincent Touzet <vincent.touzet@gmail.com>
  22. */
  23. class DoctrineTransport implements TransportInterface, SetupableTransportInterface, MessageCountAwareInterface, ListableReceiverInterface
  24. {
  25. private $connection;
  26. private $serializer;
  27. private $receiver;
  28. private $sender;
  29. public function __construct(Connection $connection, SerializerInterface $serializer)
  30. {
  31. $this->connection = $connection;
  32. $this->serializer = $serializer;
  33. }
  34. /**
  35. * {@inheritdoc}
  36. */
  37. public function get(): iterable
  38. {
  39. return ($this->receiver ?? $this->getReceiver())->get();
  40. }
  41. /**
  42. * {@inheritdoc}
  43. */
  44. public function ack(Envelope $envelope): void
  45. {
  46. ($this->receiver ?? $this->getReceiver())->ack($envelope);
  47. }
  48. /**
  49. * {@inheritdoc}
  50. */
  51. public function reject(Envelope $envelope): void
  52. {
  53. ($this->receiver ?? $this->getReceiver())->reject($envelope);
  54. }
  55. /**
  56. * {@inheritdoc}
  57. */
  58. public function getMessageCount(): int
  59. {
  60. return ($this->receiver ?? $this->getReceiver())->getMessageCount();
  61. }
  62. /**
  63. * {@inheritdoc}
  64. */
  65. public function all(int $limit = null): iterable
  66. {
  67. return ($this->receiver ?? $this->getReceiver())->all($limit);
  68. }
  69. /**
  70. * {@inheritdoc}
  71. */
  72. public function find($id): ?Envelope
  73. {
  74. return ($this->receiver ?? $this->getReceiver())->find($id);
  75. }
  76. /**
  77. * {@inheritdoc}
  78. */
  79. public function send(Envelope $envelope): Envelope
  80. {
  81. return ($this->sender ?? $this->getSender())->send($envelope);
  82. }
  83. /**
  84. * {@inheritdoc}
  85. */
  86. public function setup(): void
  87. {
  88. $this->connection->setup();
  89. }
  90. /**
  91. * Adds the Table to the Schema if this transport uses this connection.
  92. */
  93. public function configureSchema(Schema $schema, DbalConnection $forConnection): void
  94. {
  95. $this->connection->configureSchema($schema, $forConnection);
  96. }
  97. /**
  98. * Adds extra SQL if the given table was created by the Connection.
  99. *
  100. * @return string[]
  101. */
  102. public function getExtraSetupSqlForTable(Table $createdTable): array
  103. {
  104. return $this->connection->getExtraSetupSqlForTable($createdTable);
  105. }
  106. private function getReceiver(): DoctrineReceiver
  107. {
  108. return $this->receiver = new DoctrineReceiver($this->connection, $this->serializer);
  109. }
  110. private function getSender(): DoctrineSender
  111. {
  112. return $this->sender = new DoctrineSender($this->connection, $this->serializer);
  113. }
  114. }
  115. class_alias(DoctrineTransport::class, \Symfony\Component\Messenger\Transport\Doctrine\DoctrineTransport::class);