Merge remote-tracking branch 'ocramius/fix/#251-dbal-queue-safe-queue-creation' into doctrine-safe-queue-creation

Author: henrikbjorn

src/Driver/DoctrineDriver.php

@@ -3,6 +3,7 @@
 namespace Bernard\Driver;
 
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Exception\ConstraintViolationException;
 
 /**
  * Driver supporting Doctrine DBAL
@@ -33,13 +34,29 @@ public function listQueues()
     }
 
     /**
-     * {@inheritdoc}
+     * {@inheritDoc}
      */
     public function createQueue($queueName)
     {
         try {
-            $this->connection->insert('bernard_queues', ['name' => $queueName]);
-        } catch (\Exception $e) {
+            $this->connection->transactional(function () use ($queueName) {
+                $queueExistsQb = $this->connection->createQueryBuilder();
+
+                $queueExists = $queueExistsQb
+                    ->select('name')
+                    ->from('bernard_queues')
+                    ->where($queueExistsQb->expr()->eq('name', ':name'))
+                    ->setParameter('name', $queueName)
+                    ->execute();
+
+                if ($queueExists->fetch()) {
+                    // queue was already created
+                    return;
+                }
+
+                $this->connection->insert('bernard_queues', array('name' => $queueName));
+            });
+        } catch (ConstraintViolationException $ignored) {
             // Because SQL server does not support a portable INSERT ON IGNORE syntax
             // this ignores error based on primary key.
         }
@@ -100,7 +117,6 @@ public function popMessage($queueName, $duration = 5)
             usleep(10000);
         }
     }
-
     /**
      * {@inheritdoc}
      */

tests/Driver/AbstractDoctrineDriverTest.php

@@ -5,12 +5,16 @@
 use Bernard\Doctrine\MessagesSchema;
 use Bernard\Driver\DoctrineDriver;
 use Doctrine\DBAL\Platforms\MySqlPlatform;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Logging\DebugStack;
 use Doctrine\DBAL\Schema\Schema;
 
 abstract class AbstractDoctrineDriverTest extends \PHPUnit_Framework_TestCase
 {
     /**
-     * @var \Doctrine\DBAL\Connection
+     * @var Connection
      */
     private $connection;
 
@@ -72,6 +76,64 @@ public function testCreateAndRemoveQueue()
         $this->assertEquals(array('send-newsletter'), $this->driver->listQueues());
     }
 
+    public function testCreateQueueWillNotAttemptDuplicateQueueCreation()
+    {
+        $logger = new DebugStack();
+
+        $this->connection->getConfiguration()->setSQLLogger($logger);
+
+        $this->driver->createQueue('import-users');
+        $this->driver->createQueue('import-users');
+
+        self::assertCount(7, $logger->queries);
+        self::assertStringMatchesFormat('%aSTART TRANSACTION%a', $logger->queries[1]['sql']);
+        self::assertStringStartsWith('SELECT ', $logger->queries[2]['sql']);
+        self::assertStringStartsWith('INSERT ', $logger->queries[3]['sql']);
+        self::assertStringMatchesFormat('%aCOMMIT%a', $logger->queries[4]['sql']);
+        self::assertStringMatchesFormat('%aSTART TRANSACTION%a', $logger->queries[5]['sql']);
+        self::assertStringStartsWith('SELECT ', $logger->queries[6]['sql']);
+        self::assertStringMatchesFormat('%aCOMMIT%a', $logger->queries[7]['sql']);
+    }
+
+    public function testGenericExceptionsBubbleUpWhenThrownOnQueueCreation()
+    {
+        $connection   = $this->getMockBuilder('Doctrine\DBAL\Connection')->disableOriginalConstructor()->getMock();
+        $exception    = new \Exception();
+        $queryBuilder = $this->connection->createQueryBuilder();
+
+        $connection->expects(self::once())->method('transactional')->willReturnCallback('call_user_func');
+        $connection->expects(self::once())->method('insert')->willThrowException($exception);
+        $connection->expects(self::any())->method('createQueryBuilder')->willReturn($queryBuilder);
+
+        $driver = new DoctrineDriver($connection);
+
+        try {
+            $driver->createQueue('foo');
+
+            self::fail('An exception was supposed to be thrown');
+        } catch (\Exception $thrown) {
+            self::assertSame($exception, $thrown);
+        }
+    }
+
+    public function testConstraintViolationExceptionsAreIgnored()
+    {
+        $connection   = $this->getMockBuilder('Doctrine\DBAL\Connection')->disableOriginalConstructor()->getMock();
+        $queryBuilder = $this->connection->createQueryBuilder();
+        $exception    = $this
+            ->getMockBuilder('Doctrine\DBAL\Exception\ConstraintViolationException')
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $connection->expects(self::once())->method('transactional')->willReturnCallback('call_user_func');
+        $connection->expects(self::once())->method('insert')->willThrowException($exception);
+        $connection->expects(self::any())->method('createQueryBuilder')->willReturn($queryBuilder);
+
+        $driver = new DoctrineDriver($connection);
+
+        $driver->createQueue('foo');
+    }
+
     public function testPushMessageLazilyCreatesQueue()
     {
         $this->driver->pushMessage('send-newsletter', 'something');