Merge pull request #81 from php-enqueue/dsn-transport

Add ability to set transport DSN directly to default transport factory.

Author: makasim

docker-compose.yml

@@ -10,7 +10,7 @@ services:
     volumes:
       - './:/mqdev'
     environment:
-      - AMQP_DSN=amqp://rabbitmq
+      - AMQP_DSN=amqp://guest:guest@rabbitmq:5672/mqdev
       - SYMFONY__RABBITMQ__HOST=rabbitmq
       - SYMFONY__RABBITMQ__USER=guest
       - SYMFONY__RABBITMQ__PASSWORD=guest

docs/bundle/quick_tour.md

@@ -45,8 +45,7 @@ First, you have to configure a transport layer and set one to be default.
 
 enqueue:
     transport:
-        default: 'amqp'
-        amqp: "amqp://"
+        default: "amqp://"
     client: ~
 ```
 

docs/client/quick_tour.md

@@ -22,15 +22,7 @@ use Enqueue\SimpleClient\SimpleClient;
 
 include __DIR__.'/vendor/autoload.php';
 
-$client = new SimpleClient([
-    'transport' => [
-        'default' => 'amqp',
-        'amqp' => 'amqp://'
-    ],
-    'client' => [
-        'app_name' => 'plain_php',
-    ],
-]);
+$client = new SimpleClient('amqp://');
 ```
 
 ## Produce message

docs/client/rpc_call.md

@@ -1,18 +1,23 @@
 # Client. RPC call
 
 The client's [quick tour](quick_tour.md) describes how to get the client object. 
-We use you followed instructions there and have instance of `Enqueue\SimpleClient\SimpleClient` in `$client` var. 
+Here we'll use `Enqueue\SimpleClient\SimpleClient` though it is not required.
+You can get all that stuff from manually built client or get objects from a container (Symfony).
+
+The simple client could be created like this:
 
 ## The client side
 
 There is a handy class RpcClient shipped with the client component. 
-It allows you to easily send a message and wait for a reply.
+It allows you to easily perform [RPC calls](https://en.wikipedia.org/wiki/Remote_procedure_call).
+It send a message and wait for a reply.
 
 ```php
 <?php
 use Enqueue\Client\RpcClient;
+use Enqueue\SimpleClient\SimpleClient;
 
-/** @var \Enqueue\SimpleClient\SimpleClient $client */
+$client = new SimpleClient('amqp://');
 
 $rpcClient = new RpcClient($client->getProducer(), $context);
 
@@ -24,8 +29,9 @@ You can perform several requests asynchronously with `callAsync` and request rep
 ```php
 <?php
 use Enqueue\Client\RpcClient;
+use Enqueue\SimpleClient\SimpleClient;
 
-/** @var \Enqueue\SimpleClient\SimpleClient $client */
+$client = new SimpleClient('amqp://');
 
 $rpcClient = new RpcClient($client->getProducer(), $context);
 
@@ -54,10 +60,11 @@ use Enqueue\Psr\PsrContext;
 use Enqueue\Consumption\Result;
 use Enqueue\Consumption\ChainExtension;
 use Enqueue\Consumption\Extension\ReplyExtension;
+use Enqueue\SimpleClient\SimpleClient;
 
 /** @var \Enqueue\Psr\PsrContext $context */
 
-/** @var \Enqueue\SimpleClient\SimpleClient $client */
+$client = new SimpleClient('amqp://');
 
 $client->bind('greeting_topic', 'greeting_processor', function (PsrMessage $message, PsrContext $context) use (&$requestMessage) {
     echo $message->getBody();

docs/quick_tour.md

@@ -170,19 +170,11 @@ Here's an example of how you can send and consume messages.
 use Enqueue\SimpleClient\SimpleClient;
 use Enqueue\Psr\PsrMessage;
 
-$client = new SimpleClient([
-    'transport' => [
-        'default' => 'amqp',
-        'amqp' => [
-            'host'  => 'localhost',
-            'port'  => 5672,
-            'vhost' => '/',
-            'user' => 'guest',
-            'pass' => 'guest',
-        ],
-    ],
-    'client' => true,
-]);
+// composer require enqueue/amqp-ext
+$client = new SimpleClient('amqp://');
+
+// composer require enqueue/fs
+$client = new SimpleClient('file://foo/bar');
 
 $client->setupBroker();
 

pkg/amqp-ext/AmqpConnectionFactory.php

@@ -77,7 +77,9 @@ private function establishConnection()
             $config = $this->config;
             $config['login'] = $this->config['user'];
             $config['password'] = $this->config['pass'];
+
             $this->connection = new \AMQPConnection($config);
+
             $this->config['persisted'] ? $this->connection->pconnect() : $this->connection->connect();
         }
         if (false == $this->connection->isConnected()) {

pkg/enqueue-bundle/Tests/Functional/UseCasesTest.php

@@ -43,6 +43,12 @@ public function provideEnqueueConfigs()
             ],
         ]];
 
+        yield 'default_amqp_as_dsn' => [[
+            'transport' => [
+                'default' => getenv('AMQP_DSN'),
+            ],
+        ]];
+
         yield 'stomp' => [[
             'transport' => [
                 'default' => 'stomp',
@@ -97,6 +103,12 @@ public function provideEnqueueConfigs()
             ],
         ]];
 
+        yield 'default_fs_as_dsn' => [[
+            'transport' => [
+                'default' => 'file:/'.sys_get_temp_dir(),
+            ],
+        ]];
+
         yield 'dbal' => [[
             'transport' => [
                 'default' => 'dbal',

pkg/enqueue-bundle/Tests/Unit/DependencyInjection/EnqueueExtensionTest.php

@@ -7,14 +7,14 @@
 use Enqueue\Bundle\Tests\Unit\Mocks\FooTransportFactory;
 use Enqueue\Client\Producer;
 use Enqueue\Client\TraceableProducer;
-use Enqueue\Symfony\DefaultTransportFactory;
+use Enqueue\Null\NullContext;
 use Enqueue\Null\Symfony\NullTransportFactory;
+use Enqueue\Symfony\DefaultTransportFactory;
 use Enqueue\Test\ClassExtensionTrait;
-use Enqueue\Null\NullContext;
+use PHPUnit\Framework\TestCase;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\HttpKernel\DependencyInjection\Extension;
-use PHPUnit\Framework\TestCase;
 
 class EnqueueExtensionTest extends TestCase
 {
@@ -95,6 +95,30 @@ public function testShouldUseNullTransportAsDefault()
         );
     }
 
+    public function testShouldUseNullTransportAsDefaultConfiguredViaDSN()
+    {
+        $container = new ContainerBuilder();
+
+        $extension = new EnqueueExtension();
+        $extension->addTransportFactory(new NullTransportFactory());
+        $extension->addTransportFactory(new DefaultTransportFactory());
+
+        $extension->load([[
+            'transport' => [
+                'default' => 'null://',
+            ],
+        ]], $container);
+
+        self::assertEquals(
+            'enqueue.transport.default.context',
+            (string) $container->getAlias('enqueue.transport.context')
+        );
+        self::assertEquals(
+            'enqueue.transport.default_null.context',
+            (string) $container->getAlias('enqueue.transport.default.context')
+        );
+    }
+
     public function testShouldConfigureFooTransport()
     {
         $container = new ContainerBuilder();
@@ -445,7 +469,7 @@ public function testShouldAddJobQueueEntityMapping()
 
         $extension->prepend($container);
 
-        $config =  $container->getExtensionConfig('doctrine');
+        $config = $container->getExtensionConfig('doctrine');
 
         $this->assertSame(['dbal' => true], $config[1]);
         $this->assertNotEmpty($config[0]['orm']['mappings']['enqueue_job_queue']);

pkg/enqueue/Symfony/DefaultTransportFactory.php

@@ -2,8 +2,15 @@
 
 namespace Enqueue\Symfony;
 
+use Enqueue\AmqpExt\AmqpConnectionFactory;
+use Enqueue\AmqpExt\Symfony\AmqpTransportFactory;
+use Enqueue\Fs\FsConnectionFactory;
+use Enqueue\Fs\Symfony\FsTransportFactory;
+use Enqueue\Null\NullConnectionFactory;
+use Enqueue\Null\Symfony\NullTransportFactory;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
+use function Enqueue\dsn_to_connection_factory;
 
 class DefaultTransportFactory implements TransportFactoryInterface
 {
@@ -29,18 +36,30 @@ public function addConfiguration(ArrayNodeDefinition $builder)
             ->beforeNormalization()
                 ->ifString()
                 ->then(function ($v) {
-                    return ['alias' => $v];
+                    if (false === strpos($v, '://')) {
+                        return ['alias' => $v];
+                    }
+
+                    return ['dsn' => $v];
                 })
             ->end()
             ->children()
-                ->scalarNode('alias')->isRequired()->cannotBeEmpty()->end()
-        ;
+                ->scalarNode('alias')->cannotBeEmpty()->end()
+                ->scalarNode('dsn')->cannotBeEmpty()->end()
+            ;
     }
 
     public function createConnectionFactory(ContainerBuilder $container, array $config)
     {
+        if (isset($config['alias'])) {
+            $aliasId = sprintf('enqueue.transport.%s.connection_factory', $config['alias']);
+        } elseif (isset($config['dsn'])) {
+            $aliasId = $this->findFactory($config['dsn'])->createConnectionFactory($container, $config);
+        } else {
+            throw new \LogicException('Either dsn or alias option must be set.');
+        }
+
         $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
-        $aliasId = sprintf('enqueue.transport.%s.connection_factory', $config['alias']);
 
         $container->setAlias($factoryId, $aliasId);
         $container->setAlias('enqueue.transport.connection_factory', $factoryId);
@@ -53,8 +72,15 @@ public function createConnectionFactory(ContainerBuilder $container, array $conf
      */
     public function createContext(ContainerBuilder $container, array $config)
     {
+        if (isset($config['alias'])) {
+            $aliasId = sprintf('enqueue.transport.%s.context', $config['alias']);
+        } elseif (isset($config['dsn'])) {
+            $aliasId = $this->findFactory($config['dsn'])->createContext($container, $config);
+        } else {
+            throw new \LogicException('Either dsn or alias option must be set.');
+        }
+
         $contextId = sprintf('enqueue.transport.%s.context', $this->getName());
-        $aliasId = sprintf('enqueue.transport.%s.context', $config['alias']);
 
         $container->setAlias($contextId, $aliasId);
         $container->setAlias('enqueue.transport.context', $contextId);
@@ -67,8 +93,15 @@ public function createContext(ContainerBuilder $container, array $config)
      */
     public function createDriver(ContainerBuilder $container, array $config)
     {
+        if (isset($config['alias'])) {
+            $aliasId = sprintf('enqueue.client.%s.driver', $config['alias']);
+        } elseif (isset($config['dsn'])) {
+            $aliasId = $this->findFactory($config['dsn'])->createDriver($container, $config);
+        } else {
+            throw new \LogicException('Either dsn or alias option must be set.');
+        }
+
         $driverId = sprintf('enqueue.client.%s.driver', $this->getName());
-        $aliasId = sprintf('enqueue.client.%s.driver', $config['alias']);
 
         $container->setAlias($driverId, $aliasId);
         $container->setAlias('enqueue.client.driver', $driverId);
@@ -83,4 +116,33 @@ public function getName()
     {
         return $this->name;
     }
+
+    /**
+     * @param string
+     * @param mixed $dsn
+     *
+     * @return TransportFactoryInterface
+     */
+    private function findFactory($dsn)
+    {
+        $connectionFactory = dsn_to_connection_factory($dsn);
+
+        if ($connectionFactory instanceof AmqpConnectionFactory) {
+            return new AmqpTransportFactory('default_amqp');
+        }
+
+        if ($connectionFactory instanceof FsConnectionFactory) {
+            return new FsTransportFactory('default_fs');
+        }
+
+        if ($connectionFactory instanceof NullConnectionFactory) {
+            return new NullTransportFactory('default_null');
+        }
+
+        throw new \LogicException(sprintf(
+            'There is no supported transport factory for the connection factory "%s" created from DSN "%s"',
+            get_class($connectionFactory),
+            $dsn
+        ));
+    }
 }