Mongodb transport

pkg/mongodb/Client/MongodbDriver.php

@@ -42,8 +42,8 @@ class MongodbDriver implements DriverInterface
     ];
 
     /**
-     * @param MongodbContext $context
-     * @param Config $config
+     * @param MongodbContext    $context
+     * @param Config            $config
      * @param QueueMetaRegistry $queueMetaRegistry
      */
     public function __construct(MongodbContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
@@ -161,7 +161,7 @@ public function setupBroker(LoggerInterface $logger = null)
     {
         $logger = $logger ?: new NullLogger();
         $log = function ($text, ...$args) use ($logger) {
-            $logger->debug(sprintf('[MongodbDriver] ' . $text, ...$args));
+            $logger->debug(sprintf('[MongodbDriver] '.$text, ...$args));
         };
         $contextConfig = $this->context->getConfig();
         $log('Creating database and collection: "%s" "%s"', $contextConfig['dbname'], $contextConfig['collection_name']);

pkg/mongodb/Symfony/MongodbTransportFactory.php

@@ -0,0 +1,123 @@
+<?php
+
+namespace Enqueue\Mongodb\Symfony;
+
+use Enqueue\Mongodb\Client\MongodbDriver;
+use Enqueue\Mongodb\MongodbConnectionFactory;
+use Enqueue\Mongodb\MongodbContext;
+use Enqueue\Symfony\DriverFactoryInterface;
+use Enqueue\Symfony\TransportFactoryInterface;
+use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Definition;
+use Symfony\Component\DependencyInjection\Reference;
+
+class MongodbTransportFactory implements TransportFactoryInterface, DriverFactoryInterface
+{
+    /**
+     * @var string
+     */
+    private $name;
+
+    /**
+     * @param string $name
+     */
+    public function __construct($name = 'Mongodb')
+    {
+        $this->name = $name;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function addConfiguration(ArrayNodeDefinition $builder)
+    {
+        $builder
+            ->beforeNormalization()
+                ->ifString()
+                ->then(function ($v) {
+                    return ['dsn' => $v];
+                })
+            ->end()
+            ->children()
+                ->scalarNode('dsn')
+                    ->info('The Mongodb DSN. Other parameters are ignored if set')
+                ->end()
+                ->scalarNode('dbname')
+                    ->defaultValue('enqueue')
+                    ->info('Database name.')
+                ->end()
+                ->scalarNode('collection_name')
+                    ->defaultValue('enqueue')
+                    ->info('Collection')
+                ->end()
+                ->integerNode('polling_interval')
+                    ->defaultValue(1000)
+                    ->min(100)
+                    ->info('How often query for new messages.')
+                ->end()
+        ;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createConnectionFactory(ContainerBuilder $container, array $config)
+    {
+        if (false == empty($config['dsn'])) {
+            $factory = new Definition(MongodbConnectionFactory::class);
+            $factory->setArguments([$config]);
+        } else {
+            throw new \LogicException('Set "dsn" options when you want ot use Mongodb.');
+        }
+
+        $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
+        $container->setDefinition($factoryId, $factory);
+
+        return $factoryId;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createContext(ContainerBuilder $container, array $config)
+    {
+        $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
+
+        $context = new Definition(MongodbContext::class);
+        $context->setPublic(true);
+        $context->setFactory([new Reference($factoryId), 'createContext']);
+
+        $contextId = sprintf('enqueue.transport.%s.context', $this->getName());
+        $container->setDefinition($contextId, $context);
+
+        return $contextId;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createDriver(ContainerBuilder $container, array $config)
+    {
+        $driver = new Definition(MongodbDriver::class);
+        $driver->setPublic(true);
+        $driver->setArguments([
+            new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
+            new Reference('enqueue.client.config'),
+            new Reference('enqueue.client.meta.queue_meta_registry'),
+        ]);
+
+        $driverId = sprintf('enqueue.client.%s.driver', $this->getName());
+        $container->setDefinition($driverId, $driver);
+
+        return $driverId;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getName()
+    {
+        return $this->name;
+    }
+}

pkg/mongodb/Tests/MongodbConnectionFactoryTest.php

@@ -20,8 +20,6 @@ public function testCouldBeConstructedWithEmptyConfiguration()
     {
         $params = [
             'uri' => 'mongodb://127.0.0.1/',
-            'uriOptions' => [],
-            'driverOptions' => [],
         ];
 
         $factory = new MongodbConnectionFactory();

pkg/mongodb/Tests/Symfony/MongodbTransportFactoryTest.php

@@ -0,0 +1,161 @@
+<?php
+
+namespace Enqueue\Mongodb\Tests\Symfony;
+
+use Enqueue\Mongodb\Client\MongodbDriver;
+use Enqueue\Mongodb\MongodbConnectionFactory;
+use Enqueue\Mongodb\Symfony\MongodbTransportFactory;
+use Enqueue\Symfony\TransportFactoryInterface;
+use Enqueue\Test\ClassExtensionTrait;
+use Symfony\Component\Config\Definition\Builder\TreeBuilder;
+use Symfony\Component\Config\Definition\Processor;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Reference;
+
+class MongodbTransportFactoryTest extends \PHPUnit_Framework_TestCase
+{
+    use ClassExtensionTrait;
+
+    public function testShouldImplementTransportFactoryInterface()
+    {
+        $this->assertClassImplements(TransportFactoryInterface::class, MongodbTransportFactory::class);
+    }
+
+    public function testCouldBeConstructedWithDefaultName()
+    {
+        $transport = new MongodbTransportFactory();
+
+        $this->assertEquals('Mongodb', $transport->getName());
+    }
+
+    public function testCouldBeConstructedWithCustomName()
+    {
+        $transport = new MongodbTransportFactory('theCustomName');
+
+        $this->assertEquals('theCustomName', $transport->getName());
+    }
+
+    public function testShouldAllowAddConfiguration()
+    {
+        $transport = new MongodbTransportFactory();
+        $tb = new TreeBuilder();
+        $rootNode = $tb->root('foo');
+
+        $transport->addConfiguration($rootNode);
+        $processor = new Processor();
+        $config = $processor->process($tb->buildTree(), [[
+            'dsn' => 'mongodb://127.0.0.1/',
+        ]]);
+
+        $this->assertEquals([
+            'dsn' => 'mongodb://127.0.0.1/',
+            'dbname' => 'enqueue',
+            'collection_name' => 'enqueue',
+            'polling_interval' => 1000,
+        ], $config);
+    }
+
+    public function testShouldAllowAddConfigurationAsString()
+    {
+        $transport = new MongodbTransportFactory();
+        $tb = new TreeBuilder();
+        $rootNode = $tb->root('foo');
+
+        $transport->addConfiguration($rootNode);
+        $processor = new Processor();
+        $config = $processor->process($tb->buildTree(), ['mysqlDSN']);
+
+        $this->assertEquals([
+            'dsn' => 'mysqlDSN',
+            'dbname' => 'enqueue',
+            'collection_name' => 'enqueue',
+            'polling_interval' => 1000,
+        ], $config);
+    }
+
+    public function testShouldCreateMongodbConnectionFactory()
+    {
+        $container = new ContainerBuilder();
+
+        $transport = new MongodbTransportFactory();
+
+        $serviceId = $transport->createConnectionFactory($container, [
+            'dsn' => 'mysqlDSN',
+            'dbname' => 'enqueue',
+            'collection_name' => 'enqueue',
+            'polling_interval' => 1000,
+        ]);
+
+        $this->assertTrue($container->hasDefinition($serviceId));
+        $factory = $container->getDefinition($serviceId);
+        $this->assertEquals(MongodbConnectionFactory::class, $factory->getClass());
+
+        $this->assertSame([
+            'dsn' => 'mysqlDSN',
+            'dbname' => 'enqueue',
+            'collection_name' => 'enqueue',
+            'polling_interval' => 1000,
+        ], $factory->getArgument(0));
+    }
+
+    public function testShouldCreateConnectionFactoryFromDsnString()
+    {
+        $container = new ContainerBuilder();
+
+        $transport = new MongodbTransportFactory();
+
+        $serviceId = $transport->createConnectionFactory($container, [
+            'dsn' => 'theDSN',
+            'connection' => [],
+            'lazy' => true,
+            'table_name' => 'enqueue',
+            'polling_interval' => 1000,
+        ]);
+
+        $this->assertTrue($container->hasDefinition($serviceId));
+        $factory = $container->getDefinition($serviceId);
+        $this->assertEquals(MongodbConnectionFactory::class, $factory->getClass());
+        $this->assertSame('theDSN', $factory->getArgument(0)['dsn']);
+    }
+
+    public function testShouldCreateContext()
+    {
+        $container = new ContainerBuilder();
+
+        $transport = new MongodbTransportFactory();
+
+        $serviceId = $transport->createContext($container, []);
+
+        $this->assertEquals('enqueue.transport.Mongodb.context', $serviceId);
+        $this->assertTrue($container->hasDefinition($serviceId));
+
+        $context = $container->getDefinition('enqueue.transport.Mongodb.context');
+        $this->assertInstanceOf(Reference::class, $context->getFactory()[0]);
+        $this->assertEquals('enqueue.transport.Mongodb.connection_factory', (string) $context->getFactory()[0]);
+        $this->assertEquals('createContext', $context->getFactory()[1]);
+    }
+
+    public function testShouldCreateDriver()
+    {
+        $container = new ContainerBuilder();
+
+        $transport = new MongodbTransportFactory();
+
+        $serviceId = $transport->createDriver($container, []);
+
+        $this->assertEquals('enqueue.client.Mongodb.driver', $serviceId);
+        $this->assertTrue($container->hasDefinition($serviceId));
+
+        $driver = $container->getDefinition($serviceId);
+        $this->assertSame(MongodbDriver::class, $driver->getClass());
+
+        $this->assertInstanceOf(Reference::class, $driver->getArgument(0));
+        $this->assertEquals('enqueue.transport.Mongodb.context', (string) $driver->getArgument(0));
+
+        $this->assertInstanceOf(Reference::class, $driver->getArgument(1));
+        $this->assertEquals('enqueue.client.config', (string) $driver->getArgument(1));
+
+        $this->assertInstanceOf(Reference::class, $driver->getArgument(2));
+        $this->assertEquals('enqueue.client.meta.queue_meta_registry', (string) $driver->getArgument(2));
+    }
+}