Proof of concept for automatically updating Doctrine Schema from Messenger

weaverryan · weaverryan · commit c49b512ad6d9 · 2020-05-01T11:33:42.000-04:00
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/Connection.php b/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,23 @@ public function find($id): ?array
         return false === $data ? null : $this->decodeEnvelopeHeaders($data);
     }
 
+    public function configureSchema(Schema $schema): void
+    {
+        if ($schema->hasTable($this->configuration['table_name'])) {
+            return;
+        }
+
+        $this->addTableToSchema($schema);
+    }
+
+    /**
+     * @internal
+     */
+    public function getDbalConnection(): DBALConnection
+    {
+        return $this->driverConnection;
+    }
+
     private function createAvailableMessagesQueryBuilder(): QueryBuilder
     {
         $now = new \DateTime();
@@ -341,6 +359,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 +386,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
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/DoctrineTransport.php b/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,16 @@ public function setup(): void
         $this->connection->setup();
     }
 
+    public function configureSchema(Schema $schema, DbalConnection $dbalConnection): void
+    {
+        // only update the schema for the correct connection
+        if ($dbalConnection !== $this->connection->getDbalConnection()) {
+            return;
+        }
+
+        $this->connection->configureSchema($schema);
+    }
+
     private function getReceiver(): DoctrineReceiver
     {
         return $this->receiver = new DoctrineReceiver($this->connection, $this->serializer);
diff --git a/src/Symfony/Component/Messenger/Bridge/Doctrine/Transport/MessengerTransportDoctrineSchemaSubscriber.php b/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()
+    {
+        return [ToolEvents::postGenerateSchema];
+    }
+}
