[RFC][client] Add ability to send events or commands.

pkg/enqueue-bundle/DependencyInjection/Compiler/ExtractProcessorTagSubscriptionsTrait.php

@@ -2,6 +2,8 @@
 
 namespace Enqueue\Bundle\DependencyInjection\Compiler;
 
+use Enqueue\Client\CommandSubscriberInterface;
+use Enqueue\Client\Config;
 use Enqueue\Client\TopicSubscriberInterface;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Exception\ParameterNotFoundException;
@@ -43,7 +45,42 @@ protected function extractSubscriptions(ContainerBuilder $container, $processorS
         ];
 
         $data = [];
+        if (is_subclass_of($processorClass, CommandSubscriberInterface::class)) {
+            /** @var CommandSubscriberInterface $processorClass */
+            $params = $processorClass::getSubscribedCommand();
+            if (is_string($params)) {
+                if (empty($params)) {
+                    throw new \LogicException('The processor name (it is also the command name) must not be empty.');
+                }
+
+                $data[] = [
+                    'topicName' => Config::COMMAND_TOPIC,
+                    'queueName' => $defaultQueueName,
+                    'queueNameHardcoded' => false,
+                    'processorName' => $params,
+                ];
+            } elseif (is_array($params)) {
+                $params = array_replace($subscriptionPrototype, $params);
+                if (false == $processorName = $resolve($params['processorName'])) {
+                    throw new \LogicException('The processor name (it is also the command name) must not be empty.');
+                }
+
+                $data[] = [
+                    'topicName' => Config::COMMAND_TOPIC,
+                    'queueName' => $resolve($params['queueName']) ?: $defaultQueueName,
+                    'queueNameHardcoded' => $resolve($params['queueNameHardcoded']),
+                    'processorName' => $processorName,
+                ];
+            } else {
+                throw new \LogicException(sprintf(
+                    'Command subscriber configuration is invalid. "%s"',
+                    json_encode($processorClass::getSubscribedCommand())
+                ));
+            }
+        }
+
         if (is_subclass_of($processorClass, TopicSubscriberInterface::class)) {
+            /** @var TopicSubscriberInterface $processorClass */
             foreach ($processorClass::getSubscribedTopics() as $topicName => $params) {
                 if (is_string($params)) {
                     $data[] = [
@@ -68,7 +105,12 @@ protected function extractSubscriptions(ContainerBuilder $container, $processorS
                     ));
                 }
             }
-        } else {
+        }
+
+        if (false == (
+            is_subclass_of($processorClass, CommandSubscriberInterface::class) ||
+            is_subclass_of($processorClass, TopicSubscriberInterface::class)
+        )) {
             foreach ($tagAttributes as $tagAttribute) {
                 $tagAttribute = array_replace($subscriptionPrototype, $tagAttribute);
 

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

@@ -23,6 +23,12 @@ services:
     enqueue.producer:
         alias: 'enqueue.client.producer'
 
+    enqueue.client.producer_v2:
+        class: 'Enqueue\Client\ProducerV2'
+        arguments:
+            - '@enqueue.client.producer'
+            - '@enqueue.client.rpc_client'
+
     enqueue.spool_producer:
         alias: 'enqueue.client.spool_producer'
 

pkg/enqueue-bundle/Tests/Functional/App/TestCommandSubscriberProcessor.php

@@ -0,0 +1,28 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Functional\App;
+
+use Enqueue\Client\CommandSubscriberInterface;
+use Enqueue\Consumption\Result;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Psr\PsrProcessor;
+
+class TestCommandSubscriberProcessor implements PsrProcessor, CommandSubscriberInterface
+{
+    public $calls = [];
+
+    public function process(PsrMessage $message, PsrContext $context)
+    {
+        $this->calls[] = $message;
+
+        return Result::reply(
+            $context->createMessage($message->getBody().'Reply')
+        );
+    }
+
+    public static function getSubscribedCommand()
+    {
+        return 'test_command_subscriber';
+    }
+}

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

@@ -47,6 +47,11 @@ services:
         tags:
             - { name: 'kernel.event_listener', async: true, event: 'test_async', method: 'onEvent' }
 
+    test_command_subscriber_processor:
+        class: 'Enqueue\Bundle\Tests\Functional\App\TestCommandSubscriberProcessor'
+        tags:
+            - { name: 'enqueue.client.processor' }
+
     test_async_subscriber:
         class: 'Enqueue\Bundle\Tests\Functional\App\TestAsyncSubscriber'
         tags:

pkg/enqueue-bundle/Tests/Functional/Client/ProducerV2Test.php

@@ -0,0 +1,97 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Functional\Client;
+
+use Enqueue\Bundle\Tests\Functional\WebTestCase;
+use Enqueue\Client\Config;
+use Enqueue\Client\ProducerV2Interface;
+use Enqueue\Client\TraceableProducer;
+use Enqueue\Rpc\Promise;
+
+/**
+ * @group functional
+ */
+class ProducerV2Test extends WebTestCase
+{
+    public function setUp()
+    {
+        parent::setUp();
+
+        $this->container->get('enqueue.client.producer')->clearTraces();
+    }
+
+    public function tearDown()
+    {
+        parent::tearDown();
+
+        $this->container->get('enqueue.client.producer')->clearTraces();
+    }
+
+    public function testCouldBeGetFromContainerAsService()
+    {
+        $producer = $this->container->get('enqueue.client.producer_v2');
+
+        $this->assertInstanceOf(ProducerV2Interface::class, $producer);
+    }
+
+    public function testShouldSendEvent()
+    {
+        /** @var ProducerV2Interface $producer */
+        $producer = $this->container->get('enqueue.client.producer_v2');
+
+        $producer->sendEvent('theTopic', 'theMessage');
+
+        $traces = $this->getTraceableProducer()->getTopicTraces('theTopic');
+
+        $this->assertCount(1, $traces);
+        $this->assertEquals('theMessage', $traces[0]['body']);
+    }
+
+    public function testShouldSendCommandWithoutNeedForReply()
+    {
+        /** @var ProducerV2Interface $producer */
+        $producer = $this->container->get('enqueue.client.producer_v2');
+
+        $result = $producer->sendCommand('theCommand', 'theMessage', false);
+
+        $this->assertNull($result);
+
+        $traces = $this->getTraceableProducer()->getTopicTraces(Config::COMMAND_TOPIC);
+
+        $this->assertCount(1, $traces);
+        $this->assertEquals('theMessage', $traces[0]['body']);
+        $this->assertEquals([
+            'enqueue.topic_name' => Config::COMMAND_TOPIC,
+            'enqueue.processor_name' => 'theCommand',
+            'enqueue.processor_queue_name' => 'default',
+        ], $traces[0]['properties']);
+    }
+
+    public function testShouldSendCommandWithNeedForReply()
+    {
+        /** @var ProducerV2Interface $producer */
+        $producer = $this->container->get('enqueue.client.producer_v2');
+
+        $result = $producer->sendCommand('theCommand', 'theMessage', true);
+
+        $this->assertInstanceOf(Promise::class, $result);
+
+        $traces = $this->getTraceableProducer()->getTopicTraces(Config::COMMAND_TOPIC);
+
+        $this->assertCount(1, $traces);
+        $this->assertEquals('theMessage', $traces[0]['body']);
+        $this->assertEquals([
+            'enqueue.topic_name' => Config::COMMAND_TOPIC,
+            'enqueue.processor_name' => 'theCommand',
+            'enqueue.processor_queue_name' => 'default',
+        ], $traces[0]['properties']);
+    }
+
+    /**
+     * @return TraceableProducer|object
+     */
+    private function getTraceableProducer()
+    {
+        return $this->container->get('enqueue.client.producer');
+    }
+}

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

@@ -30,4 +30,16 @@ public function testShouldDisplayRegisteredQueues()
         $this->assertContains('enqueue.app.default', $display);
         $this->assertContains('enqueue.client.router_processor', $display);
     }
+
+    public function testShouldDisplayRegisteredCommand()
+    {
+        $command = $this->container->get('enqueue.client.meta.queues_command');
+
+        $tester = new CommandTester($command);
+        $tester->execute([]);
+
+        $display = $tester->getDisplay();
+
+        $this->assertContains('test_command_subscriber', $display);
+    }
 }

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

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Bundle\Tests\Functional;
 
+use Enqueue\Client\Config;
 use Enqueue\Symfony\Client\Meta\TopicsCommand;
 use Symfony\Component\Console\Tester\CommandTester;
 
@@ -29,4 +30,17 @@ public function testShouldDisplayRegisteredTopics()
         $this->assertContains('__router__', $display);
         $this->assertContains('enqueue.client.router_processor', $display);
     }
+
+    public function testShouldDisplayCommands()
+    {
+        $command = $this->container->get('enqueue.client.meta.topics_command');
+
+        $tester = new CommandTester($command);
+        $tester->execute([]);
+
+        $display = $tester->getDisplay();
+
+        $this->assertContains(Config::COMMAND_TOPIC, $display);
+        $this->assertContains('test_command_subscriber', $display);
+    }
 }

pkg/enqueue-bundle/Tests/Unit/DependencyInjection/Compiler/BuildClientRoutingPassTest.php

@@ -4,9 +4,12 @@
 
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildClientRoutingPass;
 use Enqueue\Bundle\Tests\Unit\DependencyInjection\Compiler\Mock\InvalidTopicSubscriber;
+use Enqueue\Bundle\Tests\Unit\DependencyInjection\Compiler\Mock\OnlyCommandNameSubscriber;
 use Enqueue\Bundle\Tests\Unit\DependencyInjection\Compiler\Mock\OnlyTopicNameTopicSubscriber;
+use Enqueue\Bundle\Tests\Unit\DependencyInjection\Compiler\Mock\ProcessorNameCommandSubscriber;
 use Enqueue\Bundle\Tests\Unit\DependencyInjection\Compiler\Mock\ProcessorNameTopicSubscriber;
 use Enqueue\Bundle\Tests\Unit\DependencyInjection\Compiler\Mock\QueueNameTopicSubscriber;
+use Enqueue\Client\Config;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Definition;
@@ -213,8 +216,6 @@ public function testShouldBuildRouteFromSubscriberIfQueueNameSpecified()
 
     public function testShouldThrowExceptionWhenTopicSubscriberConfigurationIsInvalid()
     {
-        $this->setExpectedException(\LogicException::class, 'Topic subscriber configuration is invalid. "[12345]"');
-
         $container = $this->createContainerBuilder();
 
         $processor = new Definition(InvalidTopicSubscriber::class);
@@ -225,8 +226,60 @@ public function testShouldThrowExceptionWhenTopicSubscriberConfigurationIsInvali
         $router->setArguments(['', '']);
         $container->setDefinition('enqueue.client.router_processor', $router);
 
+        $pass = new BuildClientRoutingPass();
+
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('Topic subscriber configuration is invalid. "[12345]"');
+
+        $pass->process($container);
+    }
+
+    public function testShouldBuildRouteFromCommandSubscriberIfOnlyCommandNameSpecified()
+    {
+        $container = $this->createContainerBuilder();
+
+        $processor = new Definition(OnlyCommandNameSubscriber::class);
+        $processor->addTag('enqueue.client.processor');
+        $container->setDefinition('processor-service-id', $processor);
+
+        $router = new Definition();
+        $router->setArguments([null, null, null]);
+        $container->setDefinition('enqueue.client.router_processor', $router);
+
+        $pass = new BuildClientRoutingPass();
+        $pass->process($container);
+
+        $expectedRoutes = [
+            Config::COMMAND_TOPIC => [
+                ['the-command-name', 'aDefaultQueueName'],
+            ],
+        ];
+
+        $this->assertEquals($expectedRoutes, $router->getArgument(1));
+    }
+
+    public function testShouldBuildRouteFromCommandSubscriberIfProcessorNameSpecified()
+    {
+        $container = $this->createContainerBuilder();
+
+        $processor = new Definition(ProcessorNameCommandSubscriber::class);
+        $processor->addTag('enqueue.client.processor');
+        $container->setDefinition('processor-service-id', $processor);
+
+        $router = new Definition();
+        $router->setArguments([null, null, null]);
+        $container->setDefinition('enqueue.client.router_processor', $router);
+
         $pass = new BuildClientRoutingPass();
         $pass->process($container);
+
+        $expectedRoutes = [
+            Config::COMMAND_TOPIC => [
+                ['the-command-name', 'the-command-queue-name'],
+            ],
+        ];
+
+        $this->assertEquals($expectedRoutes, $router->getArgument(1));
     }
 
     /**