DBAL Transport

bin/subtree-split

@@ -49,6 +49,7 @@ remote stomp git@github.com:php-enqueue/stomp.git
 remote amqp-ext git@github.com:php-enqueue/amqp-ext.git
 remote fs git@github.com:php-enqueue/fs.git
 remote redis git@github.com:php-enqueue/redis.git
+remote dbal git@github.com:php-enqueue/dbal.git
 remote enqueue-bundle git@github.com:php-enqueue/enqueue-bundle.git
 remote job-queue git@github.com:php-enqueue/job-queue.git
 remote test git@github.com:php-enqueue/test.git
@@ -59,6 +60,7 @@ split 'pkg/stomp' stomp
 split 'pkg/amqp-ext' amqp-ext
 split 'pkg/fs' fs
 split 'pkg/redis' redis
+split 'pkg/dbal' dbal
 split 'pkg/enqueue-bundle' enqueue-bundle
 split 'pkg/job-queue' job-queue
 split 'pkg/test' test

composer.json

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

docs/dbal_transport.md

@@ -0,0 +1,110 @@
+# Doctrine DBAL transport
+
+The transport uses [Doctrine DBAL](http://docs.doctrine-project.org/projects/doctrine-dbal/en/latest/) library and SQL like server as a broker. 
+It creates a table there. Pushes and pops messages to\from that table. 
+
+**Limitations** It works only in auto ack mode hence If consumer crashes the message is lost.  
+
+* [Installation](#installation)
+* [Init database](#init-database)
+* [Create context](#create-context)
+* [Send message to topic](#send-message-to-topic)
+* [Send message to queue](#send-message-to-queue)
+* [Consume message](#consume-message)
+
+## Installation
+
+```bash
+$ composer require enqueue/dbal
+```
+
+## Create context
+
+* With config (a connection is created internally):
+
+```php
+<?php
+use Enqueue\Dbal\DbalConnectionFactory;
+
+$factory = new DbalConnectionFactory([
+    'connection' => [
+        'dbname' => 'mqdev',
+        'user' => 'user',
+        'password' => 'pass',
+        'host' => 'localhost',
+        'port' => 3306,
+        'driver' => 'pdo_mysql',
+    ],
+    'table_name' => 'enqueue',
+]);
+
+$psrContext = $factory->createContext();
+```
+
+* With existing connection:
+
+```php
+<?php
+use Enqueue\Dbal\ManagerRegistryConnectionFactory;
+use Doctrine\Common\Persistence\ManagerRegistry;
+
+/** @var ManagerRegistry $registry */
+
+$factory = new ManagerRegistryConnectionFactory($registry, [
+    'connection_name' => 'default',
+]);
+
+$psrContext = $factory->createContext();
+```
+
+## Init database
+
+At first time you have to create a table where your message will live. There is a handy methods for this `createDataBaseTable` on the context.
+Please pay attention to that the database has to be created manually.
+
+```php
+<?php
+/** @var \Enqueue\Dbal\DbalContext $psrContext */
+
+$psrContext->createDataBaseTable();
+```
+
+## Send message to topic
+
+```php
+<?php
+/** @var \Enqueue\Dbal\DbalContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('aTopic');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooTopic, $message);
+```
+
+## Send message to queue 
+
+```php
+<?php
+/** @var \Enqueue\Dbal\DbalContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooQueue, $message);
+```
+
+## Consume message:
+
+```php
+<?php
+/** @var \Enqueue\Dbal\DbalContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$consumer = $psrContext->createConsumer($fooQueue);
+
+$message = $consumer->receive();
+
+// process a message
+```
+
+[back to index](index.md)

phpunit.xml.dist

@@ -37,6 +37,10 @@
             <directory>pkg/redis/Tests</directory>
         </testsuite>
 
+        <testsuite name="dbal transport">
+            <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,148 @@
+<?php
+namespace Enqueue\Dbal\Client;
+
+use Enqueue\Client\Config;
+use Enqueue\Client\DriverInterface;
+use Enqueue\Client\Message;
+use Enqueue\Client\MessagePriority;
+use Enqueue\Dbal\DbalContext;
+use Enqueue\Dbal\DbalMessage;
+use Enqueue\Psr\PsrMessage;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+class DbalDriver implements DriverInterface
+{
+    /**
+     * @var DbalContext
+     */
+    private $context;
+
+    /**
+     * @var Config
+     */
+    private $config;
+
+    /**
+     * @param DbalContext $context
+     * @param Config      $config
+     */
+    public function __construct(DbalContext $context, Config $config)
+    {
+        $this->context = $context;
+        $this->config = $config;
+    }
+
+    /**
+     * {@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());
+        $transportMessage->setReplyTo($message->getReplyTo());
+        $transportMessage->setCorrelationId($message->getCorrelationId());
+
+        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());
+        $clientMessage->setReplyTo($message->getReplyTo());
+        $clientMessage->setCorrelationId($message->getCorrelationId());
+
+        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');
+        }
+
+        $queue = $this->createQueue($this->config->getRouterQueueName());
+        $transportMessage = $this->createTransportMessage($message);
+
+        $this->context->createProducer()->send($queue, $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)
+    {
+        $logger = $logger ?: new NullLogger();
+        $log = function ($text, ...$args) use ($logger) {
+            $logger->debug(sprintf('[DbalDriver] '.$text, ...$args));
+        };
+
+        $log('Creating database table: "%s"', $this->context->getTableName());
+        $this->context->createDataBaseTable();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getConfig()
+    {
+        return $this->config;
+    }
+}