automatically updating Doctrine Schema from Messenger

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/ConnectionTest.php

@@ -16,6 +16,7 @@
 use Doctrine\DBAL\Platforms\AbstractPlatform;
 use Doctrine\DBAL\Query\QueryBuilder;
 use Doctrine\DBAL\Schema\AbstractSchemaManager;
+use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Schema\SchemaConfig;
 use Doctrine\DBAL\Schema\Synchronizer\SchemaSynchronizer;
 use PHPUnit\Framework\TestCase;
@@ -343,4 +344,37 @@ public function testFindAll()
         $this->assertEquals('{"message":"Hi again"}', $doctrineEnvelopes[1]['body']);
         $this->assertEquals(['type' => DummyMessage::class], $doctrineEnvelopes[1]['headers']);
     }
+
+    public function testConfigureSchema()
+    {
+        $driverConnection = $this->getDBALConnectionMock();
+        $schema = new Schema();
+
+        $connection = new Connection(['table_name' => 'queue_table'], $driverConnection);
+        $connection->configureSchema($schema, $driverConnection);
+        $this->assertTrue($schema->hasTable('queue_table'));
+    }
+
+    public function testConfigureSchemaDifferentDbalConnection()
+    {
+        $driverConnection = $this->getDBALConnectionMock();
+        $driverConnection2 = $this->getDBALConnectionMock();
+        $schema = new Schema();
+
+        $connection = new Connection([], $driverConnection);
+        $connection->configureSchema($schema, $driverConnection2);
+        $this->assertFalse($schema->hasTable('messenger_messages'));
+    }
+
+    public function testConfigureSchemaTableExists()
+    {
+        $driverConnection = $this->getDBALConnectionMock();
+        $schema = new Schema();
+        $schema->createTable('messenger_messages');
+
+        $connection = new Connection([], $driverConnection);
+        $connection->configureSchema($schema, $driverConnection);
+        $table = $schema->getTable('messenger_messages');
+        $this->assertEmpty($table->getColumns(), 'The table was not overwritten');
+    }
 }

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/DoctrineTransportTest.php

@@ -11,6 +11,8 @@
 
 namespace Symfony\Component\Messenger\Bridge\Doctrine\Tests\Transport;
 
+use Doctrine\DBAL\Connection as DbalConnection;
+use Doctrine\DBAL\Schema\Schema;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Bridge\Doctrine\Tests\Fixtures\DummyMessage;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;
@@ -50,6 +52,23 @@ public function testReceivesMessages()
         $this->assertSame($decodedMessage, $envelopes[0]->getMessage());
     }
 
+    public function testConfigureSchema()
+    {
+        $transport = $this->getTransport(
+            null,
+            $connection = $this->createMock(Connection::class)
+        );
+
+        $schema = new Schema();
+        $dbalConnection = $this->createMock(DbalConnection::class);
+
+        $connection->expects($this->once())
+            ->method('configureSchema')
+            ->with($schema, $dbalConnection);
+
+        $transport->configureSchema($schema, $dbalConnection);
+    }
+
     private function getTransport(SerializerInterface $serializer = null, Connection $connection = null): DoctrineTransport
     {
         $serializer = $serializer ?: $this->createMock(SerializerInterface::class);

src/Symfony/Component/Messenger/Bridge/Doctrine/Tests/Transport/MessengerTransportDoctrineSchemaSubscriberTest.php

@@ -0,0 +1,46 @@
+<?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\Tests\Transport;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\ORM\EntityManagerInterface;
+use Doctrine\ORM\Tools\Event\GenerateSchemaEventArgs;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;
+use Symfony\Component\Messenger\Bridge\Doctrine\Transport\MessengerTransportDoctrineSchemaSubscriber;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+class MessengerTransportDoctrineSchemaSubscriberTest extends TestCase
+{
+    public function testPostGenerateSchema()
+    {
+        $schema = new Schema();
+        $dbalConnection = $this->createMock(Connection::class);
+        $entityManager = $this->createMock(EntityManagerInterface::class);
+        $entityManager->expects($this->once())
+            ->method('getConnection')
+            ->willReturn($dbalConnection);
+        $event = new GenerateSchemaEventArgs($entityManager, $schema);
+
+        $doctrineTransport = $this->createMock(DoctrineTransport::class);
+        $doctrineTransport->expects($this->once())
+            ->method('configureSchema')
+            ->with($schema, $dbalConnection);
+        $otherTransport = $this->createMock(TransportInterface::class);
+        $otherTransport->expects($this->never())
+            ->method($this->anything());
+
+        $subscriber = new MessengerTransportDoctrineSchemaSubscriber([$doctrineTransport, $otherTransport]);
+        $subscriber->postGenerateSchema($event);
+    }
+}

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/Connection.php

@@ -19,6 +19,7 @@
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Schema\Synchronizer\SchemaSynchronizer;
 use Doctrine\DBAL\Schema\Synchronizer\SingleDatabaseSynchronizer;
+use Doctrine\DBAL\Schema\Table;
 use Doctrine\DBAL\Types\Type;
 use Doctrine\DBAL\Types\Types;
 use Symfony\Component\Messenger\Exception\InvalidArgumentException;
@@ -290,6 +291,20 @@ public function find($id): ?array
         return false === $data ? null : $this->decodeEnvelopeHeaders($data);
     }
 
+    public function configureSchema(Schema $schema, DBALConnection $dbalConnection): void
+    {
+        // only update the schema for this connection
+        if ($dbalConnection !== $this->driverConnection) {
+            return;
+        }
+
+        if ($schema->hasTable($this->configuration['table_name'])) {
+            return;
+        }
+
+        $this->addTableToSchema($schema);
+    }
+
     private function createAvailableMessagesQueryBuilder(): QueryBuilder
     {
         $now = new \DateTime();
@@ -341,6 +356,13 @@ private function executeQuery(string $sql, array $parameters = [], array $types
     private function getSchema(): Schema
     {
         $schema = new Schema([], [], $this->driverConnection->getSchemaManager()->createSchemaConfig());
+        $this->addTableToSchema($schema);
+
+        return $schema;
+    }
+
+    private function addTableToSchema(Schema $schema): void
+    {
         $table = $schema->createTable($this->configuration['table_name']);
         $table->addColumn('id', self::$useDeprecatedConstants ? Type::BIGINT : Types::BIGINT)
             ->setAutoincrement(true)
@@ -361,8 +383,6 @@ private function getSchema(): Schema
         $table->addIndex(['queue_name']);
         $table->addIndex(['available_at']);
         $table->addIndex(['delivered_at']);
-
-        return $schema;
     }
 
     private function decodeEnvelopeHeaders(array $doctrineEnvelope): array

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineTransport.php

@@ -11,6 +11,8 @@
 
 namespace Symfony\Component\Messenger\Bridge\Doctrine\Transport;
 
+use Doctrine\DBAL\Connection as DbalConnection;
+use Doctrine\DBAL\Schema\Schema;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
@@ -98,6 +100,11 @@ public function setup(): void
         $this->connection->setup();
     }
 
+    public function configureSchema(Schema $schema, DbalConnection $dbalConnection): void
+    {
+        $this->connection->configureSchema($schema, $dbalConnection);
+    }
+
     private function getReceiver(): DoctrineReceiver
     {
         return $this->receiver = new DoctrineReceiver($this->connection, $this->serializer);

src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/MessengerTransportDoctrineSchemaSubscriber.php

@@ -0,0 +1,47 @@
+<?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\Common\EventSubscriber;
+use Doctrine\ORM\Tools\Event\GenerateSchemaEventArgs;
+use Doctrine\ORM\Tools\ToolEvents;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+class MessengerTransportDoctrineSchemaSubscriber implements EventSubscriber
+{
+    private $transports;
+
+    /**
+     * @param iterable|TransportInterface[] $transports
+     */
+    public function __construct(iterable $transports)
+    {
+        $this->transports = $transports;
+    }
+
+    public function postGenerateSchema(GenerateSchemaEventArgs $event)
+    {
+        $dbalConnection = $event->getEntityManager()->getConnection();
+        foreach ($this->transports as $transport) {
+            if (!$transport instanceof DoctrineTransport) {
+                continue;
+            }
+
+            $transport->configureSchema($event->getSchema(), $dbalConnection);
+        }
+    }
+
+    public function getSubscribedEvents(): array
+    {
+        return [ToolEvents::postGenerateSchema];
+    }
+}