vendor/symfony/doctrine-messenger/Transport/Connection.php line 388

  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\Driver\Exception as DriverException;
  16. use Doctrine\DBAL\Driver\Result as DriverResult;
  17. use Doctrine\DBAL\Exception as DBALException;
  18. use Doctrine\DBAL\Exception\TableNotFoundException;
  19. use Doctrine\DBAL\LockMode;
  20. use Doctrine\DBAL\Platforms\MySQLPlatform;
  21. use Doctrine\DBAL\Platforms\OraclePlatform;
  22. use Doctrine\DBAL\Query\QueryBuilder;
  23. use Doctrine\DBAL\Result;
  24. use Doctrine\DBAL\Schema\AbstractSchemaManager;
  25. use Doctrine\DBAL\Schema\Comparator;
  26. use Doctrine\DBAL\Schema\Schema;
  27. use Doctrine\DBAL\Schema\SchemaDiff;
  28. use Doctrine\DBAL\Schema\Synchronizer\SchemaSynchronizer;
  29. use Doctrine\DBAL\Schema\Table;
  30. use Doctrine\DBAL\Types\Types;
  31. use Symfony\Component\Messenger\Exception\InvalidArgumentException;
  32. use Symfony\Component\Messenger\Exception\TransportException;
  33. use Symfony\Contracts\Service\ResetInterface;
  35. /**
  36.  * @internal
  37.  *
  38.  * @author Vincent Touzet <vincent.touzet@gmail.com>
  39.  * @author Kévin Dunglas <dunglas@gmail.com>
  40.  */
  41. class Connection implements ResetInterface
  42. {
  43.     protected const TABLE_OPTION_NAME = '_symfony_messenger_table_name';
  45.     protected const DEFAULT_OPTIONS = [
  46.         'table_name' => 'messenger_messages',
  47.         'queue_name' => 'default',
  48.         'redeliver_timeout' => 3600,
  49.         'auto_setup' => true,
  50.     ];
  52.     /**
  53.      * Configuration of the connection.
  54.      *
  55.      * Available options:
  56.      *
  57.      * * table_name: name of the table
  58.      * * connection: name of the Doctrine's entity manager
  59.      * * queue_name: name of the queue
  60.      * * redeliver_timeout: Timeout before redeliver messages still in handling state (i.e: delivered_at is not null and message is still in table). Default: 3600
  61.      * * auto_setup: Whether the table should be created automatically during send / get. Default: true
  62.      */
  63.     protected $configuration = [];
  64.     protected $driverConnection;
  65.     protected $queueEmptiedAt;
  66.     private ?SchemaSynchronizer $schemaSynchronizer;
  67.     private bool $autoSetup;
  69.     public function __construct(array $configuration, DBALConnection $driverConnection, SchemaSynchronizer $schemaSynchronizer = null)
  70.     {
  71.         $this->configuration = array_replace_recursive(static::DEFAULT_OPTIONS, $configuration);
  72.         $this->driverConnection = $driverConnection;
  73.         $this->schemaSynchronizer = $schemaSynchronizer;
  74.         $this->autoSetup = $this->configuration['auto_setup'];
  75.     }
  77.     public function reset()
  78.     {
  79.         $this->queueEmptiedAt = null;
  80.     }
  82.     public function getConfiguration(): array
  83.     {
  84.         return $this->configuration;
  85.     }
  87.     public static function buildConfiguration(string $dsn, array $options = []): array
  88.     {
  89.         if (false === $components = parse_url($dsn)) {
  90.             throw new InvalidArgumentException(sprintf('The given Doctrine Messenger DSN "%s" is invalid.', $dsn));
  91.         }
  93.         $query = [];
  94.         if (isset($components['query'])) {
  95.             parse_str($components['query'], $query);
  96.         }
  98.         $configuration = ['connection' => $components['host']];
  99.         $configuration += $query + $options + static::DEFAULT_OPTIONS;
  101.         $configuration['auto_setup'] = filter_var($configuration['auto_setup'], \FILTER_VALIDATE_BOOL);
  103.         // check for extra keys in options
  104.         $optionsExtraKeys = array_diff(array_keys($options), array_keys(static::DEFAULT_OPTIONS));
  105.         if (0 < \count($optionsExtraKeys)) {
  106.             throw new InvalidArgumentException(sprintf('Unknown option found: [%s]. Allowed options are [%s].', implode(', ', $optionsExtraKeys), implode(', ', array_keys(static::DEFAULT_OPTIONS))));
  107.         }
  109.         // check for extra keys in options
  110.         $queryExtraKeys = array_diff(array_keys($query), array_keys(static::DEFAULT_OPTIONS));
  111.         if (0 < \count($queryExtraKeys)) {
  112.             throw new InvalidArgumentException(sprintf('Unknown option found in DSN: [%s]. Allowed options are [%s].', implode(', ', $queryExtraKeys), implode(', ', array_keys(static::DEFAULT_OPTIONS))));
  113.         }
  115.         return $configuration;
  116.     }
  118.     /**
  119.      * @param int $delay The delay in milliseconds
  120.      *
  121.      * @return string The inserted id
  122.      *
  123.      * @throws DBALException
  124.      */
  125.     public function send(string $body, array $headers, int $delay = 0): string
  126.     {
  127.         $now = new \DateTimeImmutable();
  128.         $availableAt = $now->modify(sprintf('+%d seconds', $delay / 1000));
  130.         $queryBuilder = $this->driverConnection->createQueryBuilder()
  131.             ->insert($this->configuration['table_name'])
  132.             ->values([
  133.                 'body' => '?',
  134.                 'headers' => '?',
  135.                 'queue_name' => '?',
  136.                 'created_at' => '?',
  137.                 'available_at' => '?',
  138.             ]);
  140.         $this->executeStatement($queryBuilder->getSQL(), [
  141.             $body,
  142.             json_encode($headers),
  143.             $this->configuration['queue_name'],
  144.             $now,
  145.             $availableAt,
  146.         ], [
  147.             null,
  148.             null,
  149.             null,
  150.             Types::DATETIME_MUTABLE,
  151.             Types::DATETIME_MUTABLE,
  152.         ]);
  154.         return $this->driverConnection->lastInsertId();
  155.     }
  157.     public function get(): ?array
  158.     {
  159.         if ($this->driverConnection->getDatabasePlatform() instanceof MySQLPlatform) {
  160.             try {
  161.                 $this->driverConnection->delete($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59']);
  162.             } catch (DriverException $e) {
  163.                 // Ignore the exception
  164.             }
  165.         }
  167.         get:
  168.         $this->driverConnection->beginTransaction();
  169.         try {
  170.             $query = $this->createAvailableMessagesQueryBuilder()
  171.                 ->orderBy('available_at', 'ASC')
  172.                 ->setMaxResults(1);
  174.             if ($this->driverConnection->getDatabasePlatform() instanceof OraclePlatform) {
  175.                 $query->select('m.id');
  176.             }
  178.             // Append pessimistic write lock to FROM clause if db platform supports it
  179.             $sql = $query->getSQL();
  180.             if (($fromPart = $query->getQueryPart('from')) &&
  181.                 ($table = $fromPart[0]['table'] ?? null) &&
  182.                 ($alias = $fromPart[0]['alias'] ?? null)
  183.             ) {
  184.                 $fromClause = sprintf('%s %s', $table, $alias);
  185.                 $sql = str_replace(
  186.                     sprintf('FROM %s WHERE', $fromClause),
  187.                     sprintf('FROM %s WHERE', $this->driverConnection->getDatabasePlatform()->appendLockHint($fromClause, LockMode::PESSIMISTIC_WRITE)),
  188.                     $sql
  189.                 );
  190.             }
  192.             // Wrap the rownum query in a sub-query to allow writelocks without ORA-02014 error
  193.             if ($this->driverConnection->getDatabasePlatform() instanceof OraclePlatform) {
  194.                 $sql = $this->createQueryBuilder('w')
  195.                     ->where('w.id IN ('.str_replace('SELECT a.* FROM', 'SELECT a.id FROM', $sql).')')
  196.                     ->getSQL();
  197.             }
  199.             // use SELECT ... FOR UPDATE to lock table
  200.             $stmt = $this->executeQuery(
  201.                 $sql.' '.$this->driverConnection->getDatabasePlatform()->getWriteLockSQL(),
  202.                 $query->getParameters(),
  203.                 $query->getParameterTypes()
  204.             );
  205.             $doctrineEnvelope = $stmt instanceof Result || $stmt instanceof DriverResult ? $stmt->fetchAssociative() : $stmt->fetch();
  207.             if (false === $doctrineEnvelope) {
  208.                 $this->driverConnection->commit();
  209.                 $this->queueEmptiedAt = microtime(true) * 1000;
  211.                 return null;
  212.             }
  213.             // Postgres can "group" notifications having the same channel and payload
  214.             // We need to be sure to empty the queue before blocking again
  215.             $this->queueEmptiedAt = null;
  217.             $doctrineEnvelope = $this->decodeEnvelopeHeaders($doctrineEnvelope);
  219.             $queryBuilder = $this->driverConnection->createQueryBuilder()
  220.                 ->update($this->configuration['table_name'])
  221.                 ->set('delivered_at', '?')
  222.                 ->where('id = ?');
  223.             $now = new \DateTimeImmutable();
  224.             $this->executeStatement($queryBuilder->getSQL(), [
  225.                 $now,
  226.                 $doctrineEnvelope['id'],
  227.             ], [
  228.                 Types::DATETIME_MUTABLE,
  229.             ]);
  231.             $this->driverConnection->commit();
  233.             return $doctrineEnvelope;
  234.         } catch (\Throwable $e) {
  235.             $this->driverConnection->rollBack();
  237.             if ($this->autoSetup && $e instanceof TableNotFoundException) {
  238.                 $this->setup();
  239.                 goto get;
  240.             }
  242.             throw $e;
  243.         }
  244.     }
  246.     public function ack(string $id): bool
  247.     {
  248.         try {
  249.             if ($this->driverConnection->getDatabasePlatform() instanceof MySQLPlatform) {
  250.                 return $this->driverConnection->update($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59'], ['id' => $id]) > 0;
  251.             }
  253.             return $this->driverConnection->delete($this->configuration['table_name'], ['id' => $id]) > 0;
  254.         } catch (DBALException $exception) {
  255.             throw new TransportException($exception->getMessage(), 0, $exception);
  256.         }
  257.     }
  259.     public function reject(string $id): bool
  260.     {
  261.         try {
  262.             if ($this->driverConnection->getDatabasePlatform() instanceof MySQLPlatform) {
  263.                 return $this->driverConnection->update($this->configuration['table_name'], ['delivered_at' => '9999-12-31 23:59:59'], ['id' => $id]) > 0;
  264.             }
  266.             return $this->driverConnection->delete($this->configuration['table_name'], ['id' => $id]) > 0;
  267.         } catch (DBALException $exception) {
  268.             throw new TransportException($exception->getMessage(), 0, $exception);
  269.         }
  270.     }
  272.     public function setup(): void
  273.     {
  274.         $configuration = $this->driverConnection->getConfiguration();
  275.         $assetFilter = $configuration->getSchemaAssetsFilter();
  276.         $configuration->setSchemaAssetsFilter(null);
  277.         $this->updateSchema();
  278.         $configuration->setSchemaAssetsFilter($assetFilter);
  279.         $this->autoSetup = false;
  280.     }
  282.     public function getMessageCount(): int
  283.     {
  284.         $queryBuilder = $this->createAvailableMessagesQueryBuilder()
  285.             ->select('COUNT(m.id) AS message_count')
  286.             ->setMaxResults(1);
  288.         $stmt = $this->executeQuery($queryBuilder->getSQL(), $queryBuilder->getParameters(), $queryBuilder->getParameterTypes());
  290.         return $stmt instanceof Result || $stmt instanceof DriverResult ? $stmt->fetchOne() : $stmt->fetchColumn();
  291.     }
  293.     public function findAll(int $limit = null): array
  294.     {
  295.         $queryBuilder = $this->createAvailableMessagesQueryBuilder();
  297.         if (null !== $limit) {
  298.             $queryBuilder->setMaxResults($limit);
  299.         }
  301.         $stmt = $this->executeQuery($queryBuilder->getSQL(), $queryBuilder->getParameters(), $queryBuilder->getParameterTypes());
  302.         $data = $stmt instanceof Result || $stmt instanceof DriverResult ? $stmt->fetchAllAssociative() : $stmt->fetchAll();
  304.         return array_map(function ($doctrineEnvelope) {
  305.             return $this->decodeEnvelopeHeaders($doctrineEnvelope);
  306.         }, $data);
  307.     }
  309.     public function find(mixed $id): ?array
  310.     {
  311.         $queryBuilder = $this->createQueryBuilder()
  312.             ->where('m.id = ? and m.queue_name = ?');
  314.         $stmt = $this->executeQuery($queryBuilder->getSQL(), [$id, $this->configuration['queue_name']]);
  315.         $data = $stmt instanceof Result || $stmt instanceof DriverResult ? $stmt->fetchAssociative() : $stmt->fetch();
  317.         return false === $data ? null : $this->decodeEnvelopeHeaders($data);
  318.     }
  320.     /**
  321.      * @internal
  322.      */
  323.     public function configureSchema(Schema $schema, DBALConnection $forConnection): void
  324.     {
  325.         // only update the schema for this connection
  326.         if ($forConnection !== $this->driverConnection) {
  327.             return;
  328.         }
  330.         if ($schema->hasTable($this->configuration['table_name'])) {
  331.             return;
  332.         }
  334.         $this->addTableToSchema($schema);
  335.     }
  337.     /**
  338.      * @internal
  339.      */
  340.     public function getExtraSetupSqlForTable(Table $createdTable): array
  341.     {
  342.         return [];
  343.     }
  345.     private function createAvailableMessagesQueryBuilder(): QueryBuilder
  346.     {
  347.         $now = new \DateTimeImmutable();
  348.         $redeliverLimit = $now->modify(sprintf('-%d seconds', $this->configuration['redeliver_timeout']));
  350.         return $this->createQueryBuilder()
  351.             ->where('m.delivered_at is null OR m.delivered_at < ?')
  352.             ->andWhere('m.available_at <= ?')
  353.             ->andWhere('m.queue_name = ?')
  354.             ->setParameters([
  355.                 $redeliverLimit,
  356.                 $now,
  357.                 $this->configuration['queue_name'],
  358.             ], [
  359.                 Types::DATETIME_MUTABLE,
  360.                 Types::DATETIME_MUTABLE,
  361.             ]);
  362.     }
  364.     private function createQueryBuilder(string $alias = 'm'): QueryBuilder
  365.     {
  366.         $queryBuilder = $this->driverConnection->createQueryBuilder()
  367.             ->from($this->configuration['table_name'], $alias);
  369.         $alias .= '.';
  371.         if (!$this->driverConnection->getDatabasePlatform() instanceof OraclePlatform) {
  372.             return $queryBuilder->select($alias.'*');
  373.         }
  375.         // Oracle databases use UPPER CASE on tables and column identifiers.
  376.         // Column alias is added to force the result to be lowercase even when the actual field is all caps.
  378.         return $queryBuilder->select(str_replace(', ', ', '.$alias,
  379.             $alias.'id AS "id", body AS "body", headers AS "headers", queue_name AS "queue_name", '.
  380.             'created_at AS "created_at", available_at AS "available_at", '.
  381.             'delivered_at AS "delivered_at"'
  382.         ));
  383.     }
  385.     private function executeQuery(string $sql, array $parameters = [], array $types = [])
  386.     {
  387.         try {
  388.             $stmt = $this->driverConnection->executeQuery($sql, $parameters, $types);
  389.         } catch (TableNotFoundException $e) {
  390.             if ($this->driverConnection->isTransactionActive()) {
  391.                 throw $e;
  392.             }
  394.             // create table
  395.             if ($this->autoSetup) {
  396.                 $this->setup();
  397.             }
  398.             $stmt = $this->driverConnection->executeQuery($sql, $parameters, $types);
  399.         }
  401.         return $stmt;
  402.     }
  404.     protected function executeStatement(string $sql, array $parameters = [], array $types = [])
  405.     {
  406.         try {
  407.             if (method_exists($this->driverConnection, 'executeStatement')) {
  408.                 $stmt = $this->driverConnection->executeStatement($sql, $parameters, $types);
  409.             } else {
  410.                 $stmt = $this->driverConnection->executeUpdate($sql, $parameters, $types);
  411.             }
  412.         } catch (TableNotFoundException $e) {
  413.             if ($this->driverConnection->isTransactionActive()) {
  414.                 throw $e;
  415.             }
  417.             // create table
  418.             if ($this->autoSetup) {
  419.                 $this->setup();
  420.             }
  421.             if (method_exists($this->driverConnection, 'executeStatement')) {
  422.                 $stmt = $this->driverConnection->executeStatement($sql, $parameters, $types);
  423.             } else {
  424.                 $stmt = $this->driverConnection->executeUpdate($sql, $parameters, $types);
  425.             }
  426.         }
  428.         return $stmt;
  429.     }
  431.     private function getSchema(): Schema
  432.     {
  433.         $schema = new Schema([], [], $this->createSchemaManager()->createSchemaConfig());
  434.         $this->addTableToSchema($schema);
  436.         return $schema;
  437.     }
  439.     private function addTableToSchema(Schema $schema): void
  440.     {
  441.         $table = $schema->createTable($this->configuration['table_name']);
  442.         // add an internal option to mark that we created this & the non-namespaced table name
  443.         $table->addOption(self::TABLE_OPTION_NAME, $this->configuration['table_name']);
  444.         $table->addColumn('id', Types::BIGINT)
  445.             ->setAutoincrement(true)
  446.             ->setNotnull(true);
  447.         $table->addColumn('body', Types::TEXT)
  448.             ->setNotnull(true);
  449.         $table->addColumn('headers', Types::TEXT)
  450.             ->setNotnull(true);
  451.         $table->addColumn('queue_name', Types::STRING)
  452.             ->setLength(190) // MySQL 5.6 only supports 191 characters on an indexed column in utf8mb4 mode
  453.             ->setNotnull(true);
  454.         $table->addColumn('created_at', Types::DATETIME_MUTABLE)
  455.             ->setNotnull(true);
  456.         $table->addColumn('available_at', Types::DATETIME_MUTABLE)
  457.             ->setNotnull(true);
  458.         $table->addColumn('delivered_at', Types::DATETIME_MUTABLE)
  459.             ->setNotnull(false);
  460.         $table->setPrimaryKey(['id']);
  461.         $table->addIndex(['queue_name']);
  462.         $table->addIndex(['available_at']);
  463.         $table->addIndex(['delivered_at']);
  464.     }
  466.     private function decodeEnvelopeHeaders(array $doctrineEnvelope): array
  467.     {
  468.         $doctrineEnvelope['headers'] = json_decode($doctrineEnvelope['headers'], true);
  470.         return $doctrineEnvelope;
  471.     }
  473.     private function updateSchema(): void
  474.     {
  475.         if (null !== $this->schemaSynchronizer) {
  476.             $this->schemaSynchronizer->updateSchema($this->getSchema(), true);
  478.             return;
  479.         }
  481.         $schemaManager = $this->createSchemaManager();
  482.         $comparator = $this->createComparator($schemaManager);
  483.         $schemaDiff = $this->compareSchemas($comparator, $schemaManager->createSchema(), $this->getSchema());
  485.         foreach ($schemaDiff->toSaveSql($this->driverConnection->getDatabasePlatform()) as $sql) {
  486.             if (method_exists($this->driverConnection, 'executeStatement')) {
  487.                 $this->driverConnection->executeStatement($sql);
  488.             } else {
  489.                 $this->driverConnection->exec($sql);
  490.             }
  491.         }
  492.     }
  494.     private function createSchemaManager(): AbstractSchemaManager
  495.     {
  496.         return method_exists($this->driverConnection, 'createSchemaManager')
  497.             ? $this->driverConnection->createSchemaManager()
  498.             : $this->driverConnection->getSchemaManager();
  499.     }
  501.     private function createComparator(AbstractSchemaManager $schemaManager): Comparator
  502.     {
  503.         return method_exists($schemaManager, 'createComparator')
  504.             ? $schemaManager->createComparator()
  505.             : new Comparator();
  506.     }
  508.     private function compareSchemas(Comparator $comparator, Schema $from, Schema $to): SchemaDiff
  509.     {
  510.         return method_exists($comparator, 'compareSchemas')
  511.             ? $comparator->compareSchemas($from, $to)
  512.             : $comparator->compare($from, $to);
  513.     }
  514. }