Merge branch 'master' into 0.9

Author: makasim

DependencyInjection/Compiler/ExtractProcessorTagSubscriptionsTrait.php

@@ -83,7 +83,16 @@ protected function extractSubscriptions(ContainerBuilder $container, $processorS
 
         if (is_subclass_of($processorClass, TopicSubscriberInterface::class)) {
             /** @var TopicSubscriberInterface $processorClass */
-            foreach ($processorClass::getSubscribedTopics() as $topicName => $params) {
+            $topics = $processorClass::getSubscribedTopics();
+            if (!is_array($topics)) {
+                throw new \LogicException(sprintf(
+                    'Topic subscriber configuration is invalid for "%s::getSubscribedTopics()": expected array, got %s.',
+                    $processorClass,
+                    gettype($topics)
+                ));
+            }
+
+            foreach ($topics as $topicName => $params) {
                 if (is_string($params)) {
                     $data[] = [
                         'topicName' => $params,
@@ -102,7 +111,8 @@ protected function extractSubscriptions(ContainerBuilder $container, $processorS
                     ];
                 } else {
                     throw new \LogicException(sprintf(
-                        'Topic subscriber configuration is invalid. "%s"',
+                        'Topic subscriber configuration is invalid for "%s::getSubscribedTopics()". "%s"',
+                        $processorClass,
                         json_encode($processorClass::getSubscribedTopics())
                     ));
                 }

DependencyInjection/EnqueueExtension.php

@@ -3,7 +3,9 @@
 namespace Enqueue\Bundle\DependencyInjection;
 
 use Enqueue\AsyncEventDispatcher\DependencyInjection\AsyncEventDispatcherExtension;
+use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Client\Producer;
+use Enqueue\Client\TopicSubscriberInterface;
 use Enqueue\Client\TraceableProducer;
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\JobQueue\Job;
@@ -72,6 +74,8 @@ public function load(array $configs, ContainerBuilder $container)
         $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yml');
 
+        $this->setupAutowiringForProcessors($container);
+
         foreach ($config['transport'] as $name => $transportConfig) {
             $this->factories[$name]->createConnectionFactory($container, $transportConfig);
             $this->factories[$name]->createContext($container, $transportConfig);
@@ -220,4 +224,19 @@ private function registerJobQueueDoctrineEntityMapping(ContainerBuilder $contain
             }
         }
     }
+
+    private function setupAutowiringForProcessors(ContainerBuilder $container)
+    {
+        if (!method_exists($container, 'registerForAutoconfiguration')) {
+            return;
+        }
+
+        $container->registerForAutoconfiguration(TopicSubscriberInterface::class)
+            ->setPublic(true)
+            ->addTag('enqueue.client.processor');
+
+        $container->registerForAutoconfiguration(CommandSubscriberInterface::class)
+            ->setPublic(true)
+            ->addTag('enqueue.client.processor');
+    }
 }

EnqueueBundle.php

@@ -21,6 +21,9 @@
 use Enqueue\Fs\Symfony\FsTransportFactory;
 use Enqueue\Gps\GpsConnectionFactory;
 use Enqueue\Gps\Symfony\GpsTransportFactory;
+use Enqueue\Mongodb\Symfony\MongodbTransportFactory;
+use Enqueue\RdKafka\RdKafkaConnectionFactory;
+use Enqueue\RdKafka\Symfony\RdKafkaTransportFactory;
 use Enqueue\Redis\RedisConnectionFactory;
 use Enqueue\Redis\Symfony\RedisTransportFactory;
 use Enqueue\Sqs\SqsConnectionFactory;
@@ -104,6 +107,18 @@ class_exists(AmqpLibConnectionFactory::class)
             $extension->setTransportFactory(new MissingTransportFactory('gps', ['enqueue/gps']));
         }
 
+        if (class_exists(RdKafkaConnectionFactory::class)) {
+            $extension->setTransportFactory(new RdKafkaTransportFactory('rdkafka'));
+        } else {
+            $extension->setTransportFactory(new MissingTransportFactory('rdkafka', ['enqueue/rdkafka']));
+        }
+
+        if (class_exists(MongodbTransportFactory::class)) {
+            $extension->setTransportFactory(new MongodbTransportFactory('mongodb'));
+        } else {
+            $extension->setTransportFactory(new MissingTransportFactory('mongodb', ['enqueue/mongodb']));
+        }
+
         $container->addCompilerPass(new AsyncEventsPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
         $container->addCompilerPass(new AsyncTransformersPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
     }

Tests/Functional/App/config/custom-config.yml

@@ -1,7 +1,11 @@
 parameters:
     locale: 'en'
     secret: 'ThisTokenIsNotSoSecretChangeIt'
-
+    env(AWS_SQS_REGION): 'us-east-1'
+    env(AWS_SQS_VERSION): 'latest'
+    env(AWS_SQS_KEY): 'key'
+    env(AWS_SQS_SECRET): 'secret'
+    env(AWS_SQS_ENDPOINT): 'http://localstack:4576'
 
 framework:
     #esi:             ~
@@ -33,3 +37,15 @@ services:
         public: true
         tags:
             - { name: 'enqueue.client.processor' }
+
+    test.sqs_client:
+        public: true
+        class: Aws\Sqs\SqsClient
+        arguments:
+            -
+                endpoint: '%env(AWS_SQS_ENDPOINT)%'
+                region: '%env(AWS_SQS_REGION)%'
+                version: '%env(AWS_SQS_VERSION)%'
+                credentials:
+                    key: '%env(AWS_SQS_KEY)%'
+                    secret: '%env(AWS_SQS_SECRET)%'

Tests/Functional/UseCasesTest.php

@@ -201,11 +201,28 @@ public function provideEnqueueConfigs()
                         'key' => getenv('AWS_SQS_KEY'),
                         'secret' => getenv('AWS_SQS_SECRET'),
                         'region' => getenv('AWS_SQS_REGION'),
+                        'endpoint' => getenv('AWS_SQS_ENDPOINT'),
+                    ],
+                ],
+            ]];
+
+            yield 'sqs_client' => [[
+                'transport' => [
+                    'default' => 'sqs',
+                    'sqs' => [
+                        'client' => 'test.sqs_client',
                     ],
                 ],
             ]];
         }
 
+        yield 'mongodb_dsn' => [[
+            'transport' => [
+                'default' => 'mongodb',
+                'mongodb' => getenv('MONGO_DSN'),
+            ],
+        ]];
+
 //        yield 'gps' => [[
 //            'transport' => [
 //                'default' => 'gps',

Tests/Unit/DependencyInjection/Compiler/BuildClientRoutingPassTest.php

@@ -254,7 +254,7 @@ public function testShouldThrowExceptionWhenTopicSubscriberConfigurationIsInvali
         $pass = new BuildClientRoutingPass();
 
         $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Topic subscriber configuration is invalid. "[12345]"');
+        $this->expectExceptionMessage('Topic subscriber configuration is invalid');
 
         $pass->process($container);
     }

Tests/Unit/DependencyInjection/Compiler/BuildProcessorRegistryPassTest.php

@@ -180,7 +180,7 @@ public function testShouldBuildRouteFromSubscriberIfProcessorNameSpecified()
     public function testShouldThrowExceptionWhenTopicSubscriberConfigurationIsInvalid()
     {
         $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Topic subscriber configuration is invalid. "[12345]"');
+        $this->expectExceptionMessage('Topic subscriber configuration is invalid');
 
         $container = $this->createContainerBuilder();
 

Tests/Unit/DependencyInjection/Compiler/BuildTopicMetaSubscribersPassTest.php

@@ -302,7 +302,7 @@ public function testShouldThrowExceptionWhenTopicSubscriberConfigurationIsInvali
         $pass = new BuildTopicMetaSubscribersPass();
 
         $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Topic subscriber configuration is invalid. "[12345]"');
+        $this->expectExceptionMessage('Topic subscriber configuration is invalid');
 
         $pass->process($container);
     }

Tests/Unit/DependencyInjection/EnqueueExtensionTest.php

@@ -6,8 +6,10 @@
 use Enqueue\Bundle\DependencyInjection\EnqueueExtension;
 use Enqueue\Bundle\Tests\Unit\Mocks\FooTransportFactory;
 use Enqueue\Bundle\Tests\Unit\Mocks\TransportFactoryWithoutDriverFactory;
+use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Client\Producer;
 use Enqueue\Client\ProducerInterface;
+use Enqueue\Client\TopicSubscriberInterface;
 use Enqueue\Client\TraceableProducer;
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\JobQueue\JobRunner;
@@ -633,6 +635,31 @@ public function testShouldThrowIfPackageShouldBeInstalledToUseTransport()
         ]], $container);
     }
 
+    public function testShouldLoadProcessAutoconfigureChildDefinition()
+    {
+        if (30300 >= Kernel::VERSION_ID) {
+            $this->markTestSkipped('The autoconfigure feature is available since Symfony 3.3 version');
+        }
+
+        $container = $this->getContainerBuilder(true);
+        $extension = new EnqueueExtension();
+
+        $extension->load([[
+            'client' => [],
+            'transport' => [],
+        ]], $container);
+
+        $autoconfigured = $container->getAutoconfiguredInstanceof();
+
+        self::assertArrayHasKey(CommandSubscriberInterface::class, $autoconfigured);
+        self::assertTrue($autoconfigured[CommandSubscriberInterface::class]->hasTag('enqueue.client.processor'));
+        self::assertTrue($autoconfigured[CommandSubscriberInterface::class]->isPublic());
+
+        self::assertArrayHasKey(TopicSubscriberInterface::class, $autoconfigured);
+        self::assertTrue($autoconfigured[TopicSubscriberInterface::class]->hasTag('enqueue.client.processor'));
+        self::assertTrue($autoconfigured[TopicSubscriberInterface::class]->isPublic());
+    }
+
     /**
      * @param bool $debug
      *