Multiple transport support. Container based approach.

Author: makasim

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;
@@ -32,8 +31,7 @@ public function load(array $configs, ContainerBuilder $container): void
         $this->setupAutowiringForProcessors($container);
 
         $transportFactory = (new TransportFactory('default'));
-        $transportFactory->createConnectionFactory($container, $config['transport']);
-        $transportFactory->createContext($container, $config['transport']);
+        $transportFactory->build($container, $config['transport']);
 
         if (isset($config['client'])) {
             $loader->load('client.yml');
@@ -80,7 +78,7 @@ public function load(array $configs, ContainerBuilder $container): void
         }
 
         // todo configure queue consumer
-        $container->getDefinition(QueueConsumer::class)
+        $container->getDefinition('enqueue.transport.default.queue_consumer')
             ->replaceArgument(2, $config['consumption']['idle_timeout'])
             ->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/services.yml

@@ -1,55 +1,16 @@
-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
-        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\Symfony\Consumption\ContainerAwareConsumeMessagesCommand:
-        class: 'Enqueue\Symfony\Consumption\ContainerAwareConsumeMessagesCommand'
+    enqueue.consume_command:
+        class: 'Enqueue\Symfony\Consumption\ConfigurableConsumeCommand'
         public: true
         arguments:
-            - '@Enqueue\Consumption\QueueConsumer'
+            - '@enqueue.locator'
+            - 'enqueue.transport.%s.queue_consumer'
+            - 'enqueue.transport.%s.processor_registry'
         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
-        arguments:
-            - '@enqueue.transport.default.context'
-
-    Enqueue\Rpc\RpcClient:
-        class: 'Enqueue\Rpc\RpcClient'
-        public: true
-        arguments:
-            - '@enqueue.transport.default.context'
-            - '@enqueue.transport.rpc_factory'
-
-    # Deprecated. To be removed in 0.10.
-    enqueue.transport.rpc_client:
-        public: true
-        alias: 'Enqueue\Rpc\RpcClient'

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

@@ -23,15 +23,16 @@ services:
         alias: 'enqueue.transport.default.context'
         public: true
 
-    test_enqueue.client.default.consume_messages_command:
-        alias: 'enqueue.client.default.consume_messages_command'
+    test_enqueue.consume_command:
+        alias: 'enqueue.consume_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/UseCasesTest.php

@@ -213,39 +213,38 @@ public function testClientConsumeMessagesFromExplicitlySetQueue(array $enqueueCo
         $this->assertEquals($expectedBody, $processor->message->getBody());
     }
 
-//    /**
-//     * @dataProvider provideEnqueueConfigs
-//     */
-//    public function testTransportConsumeMessagesCommandShouldConsumeMessage(array $enqueueConfig)
-//    {
-//        $this->customSetUp($enqueueConfig);
-//
-//        if ($this->getTestQueue() instanceof StompDestination) {
-//            $this->markTestSkipped('The test fails with the exception Stomp\Exception\ErrorFrameException: Error "precondition_failed". '.
-//                'It happens because of the destination options are different from the one used while creating the dest. Nothing to do about it'
-//            );
-//        }
-//
-//        $expectedBody = __METHOD__.time();
-//
-//        $command = static::$container->get('test_enqueue.client.default.consume_messages_command');
-//        $command->setContainer(static::$container);
-//        $processor = static::$container->get('test.message.processor');
-//
-//        $this->getMessageProducer()->sendEvent(TestProcessor::TOPIC, $expectedBody);
-//
-//        $tester = new CommandTester($command);
-//        $tester->execute([
-//            '--message-limit' => 1,
-//            '--time-limit' => '+2sec',
-//            '--receive-timeout' => 1000,
-//            '--queue' => [$this->getTestQueue()->getQueueName()],
-//            'processor-service' => 'test.message.processor',
-//        ]);
-//
-//        $this->assertInstanceOf(Message::class, $processor->message);
-//        $this->assertEquals($expectedBody, $processor->message->getBody());
-//    }
+    /**
+     * @dataProvider provideEnqueueConfigs
+     */
+    public function testTransportConsumeMessagesCommandShouldConsumeMessage(array $enqueueConfig)
+    {
+        $this->customSetUp($enqueueConfig);
+
+        if ($this->getTestQueue() instanceof StompDestination) {
+            $this->markTestSkipped('The test fails with the exception Stomp\Exception\ErrorFrameException: Error "precondition_failed". '.
+                'It happens because of the destination options are different from the one used while creating the dest. Nothing to do about it'
+            );
+        }
+
+        $expectedBody = __METHOD__.time();
+
+        $command = static::$container->get('test_enqueue.consume_command');
+        $processor = static::$container->get('test.message.processor');
+
+        $this->getMessageProducer()->sendEvent(TestProcessor::TOPIC, $expectedBody);
+
+        $tester = new CommandTester($command);
+        $tester->execute([
+            '--message-limit' => 1,
+            '--time-limit' => '+2sec',
+            '--receive-timeout' => 1000,
+            'processor' => 'test.message.processor',
+            'queues' => [$this->getTestQueue()->getQueueName()],
+        ]);
+
+        $this->assertInstanceOf(Message::class, $processor->message);
+        $this->assertEquals($expectedBody, $processor->message->getBody());
+    }
 
     /**
      * @return string

pkg/enqueue/Consumption/ArrayQueueConsumerRegistry.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace Enqueue\Container;
+
+use Psr\Container\ContainerInterface;
+
+class Container implements ContainerInterface
+{
+    /**
+     * @var array
+     */
+    private $services;
+
+    public function __construct(array $services)
+    {
+        $this->services = $services;
+    }
+
+    public function get($id)
+    {
+        if (false == $this->has($id)) {
+            throw new NotFoundException(sprintf('The service "%s" not found.', $id));
+        }
+
+        return $this->services[$id];
+    }
+
+    public function has($id)
+    {
+        return array_key_exists($id, $this->services);
+    }
+}

pkg/enqueue/Consumption/QueueConsumerRegistryInterface.php

@@ -0,0 +1,9 @@
+<?php
+
+namespace Enqueue\Container;
+
+use Psr\Container\NotFoundExceptionInterface;
+
+class NotFoundException extends \InvalidArgumentException implements NotFoundExceptionInterface
+{
+}

pkg/enqueue/Container/Container.php

@@ -0,0 +1,35 @@
+<?php
+
+namespace Enqueue\Symfony\Consumption;
+
+use Enqueue\Consumption\Extension\LoggerExtension;
+use Psr\Log\NullLogger;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Logger\ConsoleLogger;
+use Symfony\Component\Console\Output\OutputInterface;
+
+trait ChooseLoggerCommandTrait
+{
+    protected function configureLoggerExtension(): void
+    {
+        $this
+            ->addOption('logger', null, InputOption::VALUE_OPTIONAL, 'A logger to be used. Could be "default", "null", "stdout".', 'default')
+        ;
+    }
+
+    protected function getLoggerExtension(InputInterface $input, OutputInterface $output): ?LoggerExtension
+    {
+        $logger = $input->getOption('logger');
+        switch ($logger) {
+            case 'null':
+                return new LoggerExtension(new NullLogger());
+            case 'stdout':
+                return new LoggerExtension(new ConsoleLogger($output));
+            case 'default':
+                return null;
+            default:
+                throw new \LogicException(sprintf('The logger "%s" is not supported', $logger));
+        }
+    }
+}