DBAL Transport

composer.json

@@ -9,6 +9,7 @@
     "enqueue/stomp": "*@dev",
     "enqueue/amqp-ext": "*@dev",
     "enqueue/fs": "*@dev",
+    "enqueue/dbal": "*@dev",
     "enqueue/enqueue-bundle": "*@dev",
     "enqueue/job-queue": "*@dev",
     "enqueue/test": "*@dev"
@@ -57,6 +58,10 @@
     {
       "type": "path",
       "url": "pkg/fs"
+    },
+    {
+      "type": "path",
+      "url": "pkg/dbal"
     }
   ]
 }

phpunit.xml.dist

@@ -33,6 +33,10 @@
             <directory>pkg/fs/Tests</directory>
         </testsuite>
 
+        <testsuite name="dbal">
+            <directory>pkg/dbal/Tests</directory>
+        </testsuite>
+
         <testsuite name="enqueue-bundle">
             <directory>pkg/enqueue-bundle/Tests</directory>
         </testsuite>

pkg/dbal/.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

pkg/dbal/.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 1
+
+language: php
+
+php:
+    - '5.6'
+    - '7.0'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install --prefer-source
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional

pkg/dbal/Client/DbalDriver.php

@@ -0,0 +1,156 @@
+<?php
+namespace Enqueue\Dbal\Client;
+
+use Enqueue\Client\Config;
+use Enqueue\Client\DriverInterface;
+use Enqueue\Client\Message;
+use Enqueue\Client\MessagePriority;
+use Enqueue\Client\Meta\QueueMetaRegistry;
+use Enqueue\Dbal\DbalContext;
+use Enqueue\Dbal\DbalDestination;
+use Enqueue\Dbal\DbalMessage;
+use Enqueue\Psr\PsrMessage;
+use Psr\Log\LoggerInterface;
+
+class DbalDriver implements DriverInterface
+{
+    /**
+     * @var DbalContext
+     */
+    private $context;
+
+    /**
+     * @var Config
+     */
+    private $config;
+
+    /**
+     * @var QueueMetaRegistry
+     */
+    private $queueMetaRegistry;
+
+    /**
+     * @param DbalContext       $context
+     * @param Config            $config
+     * @param QueueMetaRegistry $queueMetaRegistry
+     */
+    public function __construct(DbalContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
+    {
+        $this->context = $context;
+        $this->config = $config;
+        $this->queueMetaRegistry = $queueMetaRegistry;
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return DbalMessage
+     */
+    public function createTransportMessage(Message $message)
+    {
+        $properties = $message->getProperties();
+
+        $headers = $message->getHeaders();
+        $headers['content_type'] = $message->getContentType();
+
+        $transportMessage = $this->context->createMessage();
+        $transportMessage->setBody($message->getBody());
+        $transportMessage->setHeaders($headers);
+        $transportMessage->setProperties($properties);
+        $transportMessage->setMessageId($message->getMessageId());
+        $transportMessage->setTimestamp($message->getTimestamp());
+        $transportMessage->setDelay($message->getDelay());
+
+        return $transportMessage;
+    }
+
+    /**
+     * @param DbalMessage $message
+     *
+     * {@inheritdoc}
+     */
+    public function createClientMessage(PsrMessage $message)
+    {
+        $clientMessage = new Message();
+
+        $clientMessage->setBody($message->getBody());
+        $clientMessage->setHeaders($message->getHeaders());
+        $clientMessage->setProperties($message->getProperties());
+
+        $clientMessage->setContentType($message->getHeader('content_type'));
+        $clientMessage->setMessageId($message->getMessageId());
+        $clientMessage->setTimestamp($message->getTimestamp());
+        $clientMessage->setPriority(MessagePriority::NORMAL);
+        $clientMessage->setDelay($message->getDelay());
+
+        return $clientMessage;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function sendToRouter(Message $message)
+    {
+        if (false == $message->getProperty(Config::PARAMETER_TOPIC_NAME)) {
+            throw new \LogicException('Topic name parameter is required but is not set');
+        }
+
+        $topic = $this->createRouterTopic();
+        $transportMessage = $this->createTransportMessage($message);
+
+        $this->context->createProducer()->send($topic, $transportMessage);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function sendToProcessor(Message $message)
+    {
+        if (false == $message->getProperty(Config::PARAMETER_PROCESSOR_NAME)) {
+            throw new \LogicException('Processor name parameter is required but is not set');
+        }
+
+        if (false == $queueName = $message->getProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME)) {
+            throw new \LogicException('Queue name parameter is required but is not set');
+        }
+
+        $transportMessage = $this->createTransportMessage($message);
+        $destination = $this->createQueue($queueName);
+
+        $this->context->createProducer()->send($destination, $transportMessage);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createQueue($queueName)
+    {
+        return $this->context->createQueue($this->config->createTransportQueueName($queueName));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function setupBroker(LoggerInterface $logger = null)
+    {
+        // TODO: Implement setupBroker() method.
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getConfig()
+    {
+        return $this->config;
+    }
+
+    /**
+     * @return DbalDestination
+     */
+    private function createRouterTopic()
+    {
+        return $this->context->createTopic(
+            $this->config->createTransportQueueName($this->config->getRouterTopicName())
+        );
+    }
+}

pkg/dbal/DbalConnectionFactory.php

@@ -0,0 +1,70 @@
+<?php
+namespace Enqueue\Dbal;
+
+use Doctrine\Common\Persistence\ManagerRegistry;
+use Doctrine\DBAL\Connection;
+use Enqueue\Psr\PsrConnectionFactory;
+
+class DbalConnectionFactory implements PsrConnectionFactory
+{
+    /**
+     * @var ManagerRegistry
+     */
+    private $registry;
+
+    /**
+     * @var array
+     */
+    private $config;
+
+    /**
+     * $config = [
+     *   'connectionName' => Dbal connection name.
+     *   'tableName'  => Database table name.
+     *   'pollingInterval' => msec How often query for new messages
+     *   'lazy' => bool Use lazy database connection
+     * ].
+     *
+     * @param $config
+     */
+    public function __construct(ManagerRegistry $registry, array $config = [])
+    {
+        $this->config = array_replace([
+            'connectionName' => null,
+            'lazy' => true,
+        ], $config);
+
+        $this->registry = $registry;
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return DbalContext
+     */
+    public function createContext()
+    {
+        if ($this->config['lazy']) {
+            return new DbalContext(function () {
+                return $this->establishConnection();
+            }, $this->config);
+        }
+
+        return new DbalContext($this->establishConnection(), $this->config);
+    }
+
+    /**
+     * @return Connection
+     */
+    private function establishConnection()
+    {
+        return $this->registry->getConnection($this->config['connectionName']);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function close()
+    {
+    }
+}