[consumption] Add ability to consume from multiple transports.

composer.json

@@ -18,6 +18,7 @@
     "doctrine/dbal": "~2.5",
     "ramsey/uuid": "^2|^3.5",
     "psr/log": "^1",
+    "psr/container": "^1",
     "symfony/event-dispatcher": "4.0.*",
     "makasim/temp-file": "^0.2",
     "google/cloud-pubsub": "^0.6.1|^1.0",

pkg/enqueue-bundle/DependencyInjection/Configuration.php

@@ -26,8 +26,13 @@ public function getConfigTreeBuilder(): TreeBuilder
                 return ['transport' => ['dsn' => 'null:']];
             });
 
+        $transportFactory = new TransportFactory('default');
+
         $transportNode = $rootNode->children()->arrayNode('transport');
-        (new TransportFactory('default'))->addConfiguration($transportNode);
+        $transportFactory->addTransportConfiguration($transportNode);
+
+        $consumptionNode = $rootNode->children()->arrayNode('consumption');
+        $transportFactory->addQueueConsumerConfiguration($consumptionNode);
 
         $rootNode->children()
             ->arrayNode('client')->children()
@@ -40,18 +45,6 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->scalarNode('default_processor_queue')->defaultValue('default')->cannotBeEmpty()->end()
                 ->integerNode('redelivered_delay_time')->min(0)->defaultValue(0)->end()
             ->end()->end()
-            ->arrayNode('consumption')->addDefaultsIfNotSet()->children()
-                ->integerNode('idle_timeout')
-                    ->min(0)
-                    ->defaultValue(0)
-                    ->info('the time in milliseconds queue consumer waits if no message received')
-                ->end()
-                ->integerNode('receive_timeout')
-                    ->min(0)
-                    ->defaultValue(100)
-                    ->info('the time in milliseconds queue consumer waits for a message (100 ms by default)')
-                ->end()
-            ->end()->end()
             ->booleanNode('job')->defaultFalse()->end()
             ->arrayNode('async_events')
                 ->addDefaultsIfNotSet()

pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php

@@ -7,7 +7,6 @@
 use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Client\TopicSubscriberInterface;
 use Enqueue\Client\TraceableProducer;
-use Enqueue\Consumption\QueueConsumer;
 use Enqueue\JobQueue\Job;
 use Enqueue\Symfony\DependencyInjection\ClientFactory;
 use Enqueue\Symfony\DependencyInjection\TransportFactory;
@@ -29,13 +28,15 @@ public function load(array $configs, ContainerBuilder $container): void
         $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yml');
 
-        $this->setupAutowiringForProcessors($container);
-
         $transportFactory = (new TransportFactory('default'));
-        $transportFactory->createConnectionFactory($container, $config['transport']);
-        $transportFactory->createContext($container, $config['transport']);
+        $transportFactory->buildConnectionFactory($container, $config['transport']);
+        $transportFactory->buildContext($container, []);
+        $transportFactory->buildQueueConsumer($container, $config['consumption']);
+        $transportFactory->buildRpcClient($container, []);
 
         if (isset($config['client'])) {
+            $this->setupAutowiringForProcessors($container);
+
             $loader->load('client.yml');
             $loader->load('extensions/flush_spool_producer_extension.yml');
             $loader->load('extensions/exclusive_command_extension.yml');
@@ -77,17 +78,20 @@ public function load(array $configs, ContainerBuilder $container): void
                     ->replaceArgument(1, $config['client']['redelivered_delay_time'])
                 ;
             }
-        }
 
-        // todo configure queue consumer
-        $container->getDefinition(QueueConsumer::class)
-            ->replaceArgument(2, $config['consumption']['idle_timeout'])
-            ->replaceArgument(3, $config['consumption']['receive_timeout'])
-        ;
+            $locatorId = 'enqueue.locator';
+            if ($container->hasDefinition($locatorId)) {
+                $locator = $container->getDefinition($locatorId);
+                $locator->replaceArgument(0, array_replace($locator->getArgument(0), [
+                    'enqueue.client.default.queue_consumer' => new Reference('enqueue.client.default.queue_consumer'),
+                    'enqueue.client.default.driver' => new Reference('enqueue.client.default.driver'),
+                    'enqueue.client.default.delegate_processor' => new Reference('enqueue.client.default.delegate_processor'),
+                    'enqueue.client.default.producer' => new Reference('enqueue.client.default.producer'),
+                ]));
+            }
 
-        if ($container->hasDefinition('enqueue.client.default.queue_consumer')) {
             $container->getDefinition('enqueue.client.default.queue_consumer')
-                ->replaceArgument(2, $config['consumption']['idle_timeout'])
+                ->replaceArgument(2, $config['consumption']['idle_time'])
                 ->replaceArgument(3, $config['consumption']['receive_timeout'])
             ;
         }

pkg/enqueue-bundle/EnqueueBundle.php

@@ -7,11 +7,13 @@
 use Enqueue\AsyncEventDispatcher\DependencyInjection\AsyncTransformersPass;
 use Enqueue\Symfony\Client\DependencyInjection\AnalyzeRouteCollectionPass;
 use Enqueue\Symfony\Client\DependencyInjection\BuildClientExtensionsPass;
-use Enqueue\Symfony\Client\DependencyInjection\BuildCommandSubscriberRoutesPass;
-use Enqueue\Symfony\Client\DependencyInjection\BuildConsumptionExtensionsPass;
-use Enqueue\Symfony\Client\DependencyInjection\BuildProcessorRegistryPass;
-use Enqueue\Symfony\Client\DependencyInjection\BuildProcessorRoutesPass;
-use Enqueue\Symfony\Client\DependencyInjection\BuildTopicSubscriberRoutesPass;
+use Enqueue\Symfony\Client\DependencyInjection\BuildCommandSubscriberRoutesPass as BuildClientCommandSubscriberRoutesPass;
+use Enqueue\Symfony\Client\DependencyInjection\BuildConsumptionExtensionsPass as BuildClientConsumptionExtensionsPass;
+use Enqueue\Symfony\Client\DependencyInjection\BuildProcessorRegistryPass as BuildClientProcessorRegistryPass;
+use Enqueue\Symfony\Client\DependencyInjection\BuildProcessorRoutesPass as BuildClientProcessorRoutesPass;
+use Enqueue\Symfony\Client\DependencyInjection\BuildTopicSubscriberRoutesPass as BuildClientTopicSubscriberRoutesPass;
+use Enqueue\Symfony\DependencyInjection\BuildConsumptionExtensionsPass;
+use Enqueue\Symfony\DependencyInjection\BuildProcessorRegistryPass;
 use Symfony\Component\DependencyInjection\Compiler\PassConfig;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\HttpKernel\Bundle\Bundle;
@@ -20,13 +22,18 @@ class EnqueueBundle extends Bundle
 {
     public function build(ContainerBuilder $container): void
     {
+        //transport passes
         $container->addCompilerPass(new BuildConsumptionExtensionsPass('default'));
+        $container->addCompilerPass(new BuildProcessorRegistryPass('default'));
+
+        //client passes
+        $container->addCompilerPass(new BuildClientConsumptionExtensionsPass('default'));
         $container->addCompilerPass(new BuildClientExtensionsPass('default'));
-        $container->addCompilerPass(new BuildTopicSubscriberRoutesPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
-        $container->addCompilerPass(new BuildCommandSubscriberRoutesPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
-        $container->addCompilerPass(new BuildProcessorRoutesPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
+        $container->addCompilerPass(new BuildClientTopicSubscriberRoutesPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
+        $container->addCompilerPass(new BuildClientCommandSubscriberRoutesPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
+        $container->addCompilerPass(new BuildClientProcessorRoutesPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
         $container->addCompilerPass(new AnalyzeRouteCollectionPass('default'), PassConfig::TYPE_BEFORE_OPTIMIZATION, 30);
-        $container->addCompilerPass(new BuildProcessorRegistryPass('default'));
+        $container->addCompilerPass(new BuildClientProcessorRegistryPass('default'));
 
         if (class_exists(AsyncEventDispatcherExtension::class)) {
             $container->addCompilerPass(new AsyncEventsPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);

pkg/enqueue-bundle/Resources/config/client.yml

@@ -1,5 +1,6 @@
 services:
-    enqueue.client.default.context: 
+services:
+    enqueue.client.default.context:
         class: 'Interop\Queue\Context'
         factory: ['@enqueue.client.default.driver', 'getContext']
 
@@ -48,7 +49,7 @@ services:
             - '@enqueue.client.default.driver'
 
     enqueue.client.default.processor_registry:
-        class: 'Enqueue\Symfony\Client\ContainerProcessorRegistry'
+        class: 'Enqueue\Symfony\ContainerProcessorRegistry'
 
     enqueue.client.default.delegate_processor:
         class: 'Enqueue\Client\DelegateProcessor'
@@ -75,37 +76,6 @@ services:
         arguments:
             - []
 
-    enqueue.client.default.consume_messages_command:
-        class: 'Enqueue\Symfony\Client\ConsumeMessagesCommand'
-        arguments:
-            - '@enqueue.client.default.queue_consumer'
-            - '@enqueue.client.default.delegate_processor'
-            - '@enqueue.client.default.driver'
-        tags:
-            - { name: 'console.command' }
-
-    enqueue.client.default.produce_message_command:
-        class: 'Enqueue\Symfony\Client\ProduceMessageCommand'
-        arguments:
-            - '@enqueue.client.default.producer'
-        tags:
-            - { name: 'console.command' }
-
-    enqueue.client.default.setup_broker_command:
-        class: 'Enqueue\Symfony\Client\SetupBrokerCommand'
-        arguments:
-            - '@enqueue.client.default.driver'
-        tags:
-            - { name: 'console.command' }
-
-    enqueue.client.default.routes_command:
-        class: 'Enqueue\Symfony\Client\RoutesCommand'
-        arguments:
-            - '@enqueue.client.default.config'
-            - '@enqueue.client.default.route_collection'
-        tags:
-            - { name: 'console.command' }
-
     # todo
     enqueue.profiler.message_queue_collector:
         class: 'Enqueue\Bundle\Profiler\MessageQueueCollector'

pkg/enqueue-bundle/Resources/config/services.yml

@@ -1,55 +1,49 @@
-parameters:
-  enqueue.queue_consumer.default_idle_time: 0
-  enqueue.queue_consumer.default_receive_timeout: 10000
-
 services:
-    enqueue.consumption.extensions:
-        class: 'Enqueue\Consumption\ChainExtension'
-        public: false
+    enqueue.locator:
+        class: 'Symfony\Component\DependencyInjection\ServiceLocator'
         arguments:
             - []
+        tags: ['container.service_locator']
 
-    Enqueue\Consumption\QueueConsumer:
-        class: 'Enqueue\Consumption\QueueConsumer'
-        public: true
+    enqueue.transport.consume_command:
+        class: 'Enqueue\Symfony\Consumption\ConfigurableConsumeCommand'
         arguments:
-            - '@enqueue.transport.default.context'
-            - '@enqueue.consumption.extensions'
-            - '%enqueue.queue_consumer.default_idle_time%'
-            - '%enqueue.queue_consumer.default_receive_timeout%'
-
-    # Deprecated. To be removed in 0.10.
-    enqueue.consumption.queue_consumer:
-        public: true
-        alias: 'Enqueue\Consumption\QueueConsumer'
+            - '@enqueue.locator'
+            - 'enqueue.transport.%s.queue_consumer'
+            - 'enqueue.transport.%s.processor_registry'
+        tags:
+            - { name: 'console.command' }
 
-    Enqueue\Symfony\Consumption\ContainerAwareConsumeMessagesCommand:
-        class: 'Enqueue\Symfony\Consumption\ContainerAwareConsumeMessagesCommand'
-        public: true
+    enqueue.client.consume_command:
+        class: 'Enqueue\Symfony\Client\ConsumeCommand'
         arguments:
-            - '@Enqueue\Consumption\QueueConsumer'
+            - '@enqueue.locator'
+            - 'enqueue.client.%s.queue_consumer'
+            - 'enqueue.client.%s.driver'
+            - 'enqueue.client.%s.delegate_processor'
         tags:
             - { name: 'console.command' }
 
-    # Deprecated. To be removed in 0.10.
-    enqueue.command.consume_messages:
-        public: true
-        alias: 'Enqueue\Symfony\Consumption\ContainerAwareConsumeMessagesCommand'
-
-    enqueue.transport.rpc_factory:
-        class: 'Enqueue\Rpc\RpcFactory'
-        public: false
+    enqueue.client.produce_command:
+        class: 'Enqueue\Symfony\Client\ProduceCommand'
         arguments:
-            - '@enqueue.transport.default.context'
+            - '@enqueue.locator'
+            - 'enqueue.client.%s.producer'
+        tags:
+            - { name: 'console.command' }
 
-    Enqueue\Rpc\RpcClient:
-        class: 'Enqueue\Rpc\RpcClient'
-        public: true
+    enqueue.client.setup_broker_command:
+        class: 'Enqueue\Symfony\Client\SetupBrokerCommand'
         arguments:
-            - '@enqueue.transport.default.context'
-            - '@enqueue.transport.rpc_factory'
+            - '@enqueue.locator'
+            - 'enqueue.client.%s.driver'
+        tags:
+            - { name: 'console.command' }
 
-    # Deprecated. To be removed in 0.10.
-    enqueue.transport.rpc_client:
-        public: true
-        alias: 'Enqueue\Rpc\RpcClient'
+    enqueue.client.routes_command:
+        class: 'Enqueue\Symfony\Client\RoutesCommand'
+        arguments:
+            - '@enqueue.locator'
+            - 'enqueue.client.%s.driver'
+        tags:
+            - { name: 'console.command' }

pkg/enqueue-bundle/Tests/Functional/App/config/config.yml

@@ -34,6 +34,18 @@ services:
         alias: 'enqueue.client.default.traceable_producer'
         public: true
 
+    test_enqueue.transport.default.queue_consumer:
+        alias: 'enqueue.transport.default.queue_consumer'
+        public: true
+
+    test_enqueue.client.default.queue_consumer:
+        alias: 'enqueue.client.default.queue_consumer'
+        public: true
+
+    test_enqueue.transport.default.rpc_client:
+        alias: 'enqueue.transport.default.rpc_client'
+        public: true
+
     test_enqueue.client.default.producer:
         alias: 'enqueue.client.default.producer'
         public: true
@@ -50,12 +62,12 @@ services:
         alias: 'enqueue.transport.default.context'
         public: true
 
-    test_enqueue.client.default.consume_messages_command:
-        alias: 'enqueue.client.default.consume_messages_command'
+    test_enqueue.client.consume_command:
+        alias: 'enqueue.client.consume_command'
         public: true
 
-    test.enqueue.client.default.routes_command:
-        alias: 'enqueue.client.default.routes_command'
+    test.enqueue.client.routes_command:
+        alias: 'enqueue.client.routes_command'
         public: true
 
     test_async_listener:

pkg/enqueue-bundle/Tests/Functional/App/config/custom-config.yml

@@ -23,15 +23,28 @@ services:
         alias: 'enqueue.transport.default.context'
         public: true
 
-    test_enqueue.client.default.consume_messages_command:
-        alias: 'enqueue.client.default.consume_messages_command'
+    test_enqueue.transport.consume_command:
+        alias: 'enqueue.transport.consume_command'
+        public: true
+
+    test_enqueue.client.consume_command:
+        alias: 'enqueue.client.consume_command'
+        public: true
+
+    test_enqueue.client.produce_command:
+        alias: 'enqueue.client.produce_command'
+        public: true
+
+    test_enqueue.client.setup_broker_command:
+        alias: 'enqueue.client.setup_broker_command'
         public: true
 
     test.message.processor:
         class: 'Enqueue\Bundle\Tests\Functional\TestProcessor'
         public: true
         tags:
             - { name: 'enqueue.topic_subscriber', client: 'default' }
+            - { name: 'enqueue.transport.processor', transport: 'default' }
 
     test.message.command_processor:
         class: 'Enqueue\Bundle\Tests\Functional\TestCommandProcessor'

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

@@ -11,8 +11,10 @@ class QueueConsumerTest extends WebTestCase
 {
     public function testCouldBeGetFromContainerAsService()
     {
-        $queueConsumer = static::$container->get(QueueConsumer::class);
+        $queueConsumer = static::$container->get('test_enqueue.client.default.queue_consumer');
+        $this->assertInstanceOf(QueueConsumer::class, $queueConsumer);
 
+        $queueConsumer = static::$container->get('test_enqueue.transport.default.queue_consumer');
         $this->assertInstanceOf(QueueConsumer::class, $queueConsumer);
     }
 }

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

@@ -12,15 +12,15 @@ class RoutesCommandTest extends WebTestCase
 {
     public function testCouldBeGetFromContainerAsService()
     {
-        $command = static::$container->get('test.enqueue.client.default.routes_command');
+        $command = static::$container->get('test.enqueue.client.routes_command');
 
         $this->assertInstanceOf(RoutesCommand::class, $command);
     }
 
     public function testShouldDisplayRegisteredTopics()
     {
         /** @var RoutesCommand $command */
-        $command = static::$container->get('test.enqueue.client.default.routes_command');
+        $command = static::$container->get('test.enqueue.client.routes_command');
 
         $tester = new CommandTester($command);
         $tester->execute([]);
@@ -36,7 +36,7 @@ public function testShouldDisplayRegisteredTopics()
     public function testShouldDisplayCommands()
     {
         /** @var RoutesCommand $command */
-        $command = static::$container->get('test.enqueue.client.default.routes_command');
+        $command = static::$container->get('test.enqueue.client.routes_command');
 
         $tester = new CommandTester($command);
         $tester->execute([]);