Merge pull request #534 from php-enqueue/client-impr-topics-commands-configuration

[client] Introduce routes. Foundation for multi transport support.

Author: makasim

docs/client/quick_tour.md

@@ -89,7 +89,7 @@ use Interop\Queue\PsrProcessor;
 
 /** @var \Enqueue\SimpleClient\SimpleClient $client */
 
-$client->bind('a_bar_topic', 'a_processor_name', function(PsrMessage $psrMessage) {
+$client->bindTopic('a_bar_topic', function(PsrMessage $psrMessage) {
     // processing logic here
 
     return PsrProcessor::ACK;

docs/client/rpc_call.md

@@ -19,15 +19,14 @@ use Interop\Queue\PsrContext;
 use Enqueue\Consumption\Result;
 use Enqueue\Consumption\ChainExtension;
 use Enqueue\Consumption\Extension\ReplyExtension;
-use Enqueue\Client\Config;
 use Enqueue\SimpleClient\SimpleClient;
 
 /** @var \Interop\Queue\PsrContext $context */
 
 // composer require enqueue/amqp-ext # or enqueue/amqp-bunny, enqueue/amqp-lib
 $client = new SimpleClient('amqp:');
 
-$client->bind(Config::COMMAND_TOPIC, 'square', function (PsrMessage $message, PsrContext $context) use (&$requestMessage) {
+$client->bindCommand('square', function (PsrMessage $message, PsrContext $context) use (&$requestMessage) {
     $number = (int) $message->getBody();
 
     return Result::reply($context->createMessage($number ^ 2));

docs/laravel/quick_tour.md

@@ -72,7 +72,7 @@ use Interop\Queue\PsrMessage;
 use Interop\Queue\PsrProcessor;
 
 $app->resolving(SimpleClient::class, function (SimpleClient $client, $app) {
-    $client->bind('enqueue_test', 'a_processor', function(PsrMessage $message) {
+    $client->bindTopic('enqueue_test', function(PsrMessage $message) {
         // do stuff here
 
         return PsrProcessor::ACK;

docs/quick_tour.md

@@ -175,17 +175,16 @@ $client = new SimpleClient('amqp:');
 
 // composer require enqueue/fs
 $client = new SimpleClient('file://foo/bar');
-
-$client->setupBroker();
-
-$client->sendEvent('a_foo_topic', 'message');
-
-$client->bind('a_foo_topic', 'fooProcessor', function(PsrMessage $message) {
+$client->bindTopic('a_foo_topic', function(PsrMessage $message) {
     echo $message->getBody().PHP_EOL;
 
     // your event processor logic here
 });
 
+$client->setupBroker();
+
+$client->sendEvent('a_foo_topic', 'message');
+
 // this is a blocking call, it'll consume message until it is interrupted 
 $client->consume();
 ```
@@ -207,18 +206,18 @@ $client = new SimpleClient('amqp:');
 // composer require enqueue/fs
 //$client = new SimpleClient('file://foo/bar');
 
-$client->setupBroker();
-
-$client->bind(Config::COMMAND_TOPIC, 'bar_command', function(PsrMessage $message) {
+$client->bindCommand('bar_command', function(PsrMessage $message) {
     // your bar command processor logic here
 });
 
-$client->bind(Config::COMMAND_TOPIC, 'baz_reply_command', function(PsrMessage $message, PsrContext $context) {
+$client->bindCommand('baz_reply_command', function(PsrMessage $message, PsrContext $context) {
     // your baz reply command processor logic here
 
     return Result::reply($context->createMessage('theReplyBody'));
 });
 
+$client->setupBroker();
+
 // It is sent to one consumer.  
 $client->sendCommand('bar_command', 'aMessageData');
 

pkg/async-command/Resources/config/services.yml

@@ -1,8 +1,7 @@
 services:
     enqueue.async_command.run_command_processor:
         class: 'Enqueue\AsyncCommand\RunCommandProcessor'
-        public: public
         arguments:
             - '%kernel.project_dir%'
         tags:
-            - { name: 'enqueue.client.processor' }
+            - { name: 'enqueue.command_subscriber', client: 'default' }

pkg/async-command/RunCommandProcessor.php

@@ -45,9 +45,9 @@ public function process(PsrMessage $message, PsrContext $context): Result
     public static function getSubscribedCommand(): array
     {
         return [
-            'processorName' => Commands::RUN_COMMAND,
-            'queueName' => Commands::RUN_COMMAND,
-            'queueNameHardcoded' => true,
+            'command' => Commands::RUN_COMMAND,
+            'queue' => Commands::RUN_COMMAND,
+            'prefix_queue' => false,
             'exclusive' => true,
         ];
     }

pkg/async-command/Tests/RunCommandProcessorTest.php

@@ -43,9 +43,9 @@ public function testShouldSubscribeOnRunCommand()
         $subscription = RunCommandProcessor::getSubscribedCommand();
 
         $this->assertSame([
-            'processorName' => Commands::RUN_COMMAND,
-            'queueName' => Commands::RUN_COMMAND,
-            'queueNameHardcoded' => true,
+            'command' => Commands::RUN_COMMAND,
+            'queue' => Commands::RUN_COMMAND,
+            'prefix_queue' => false,
             'exclusive' => true,
         ], $subscription);
     }

pkg/async-event-dispatcher/AsyncProcessor.php

@@ -2,13 +2,14 @@
 
 namespace Enqueue\AsyncEventDispatcher;
 
+use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Consumption\Result;
 use Interop\Queue\PsrContext;
 use Interop\Queue\PsrMessage;
 use Interop\Queue\PsrProcessor;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
-class AsyncProcessor implements PsrProcessor
+class AsyncProcessor implements PsrProcessor, CommandSubscriberInterface
 {
     /**
      * @var Registry
@@ -20,10 +21,6 @@ class AsyncProcessor implements PsrProcessor
      */
     private $dispatcher;
 
-    /**
-     * @param Registry                 $registry
-     * @param EventDispatcherInterface $dispatcher
-     */
     public function __construct(Registry $registry, EventDispatcherInterface $dispatcher)
     {
         $this->registry = $registry;
@@ -39,9 +36,6 @@ public function __construct(Registry $registry, EventDispatcherInterface $dispat
         $this->dispatcher = $dispatcher;
     }
 
-    /**
-     * {@inheritdoc}
-     */
     public function process(PsrMessage $message, PsrContext $context)
     {
         if (false == $eventName = $message->getProperty('event_name')) {
@@ -57,4 +51,9 @@ public function process(PsrMessage $message, PsrContext $context)
 
         return self::ACK;
     }
+
+    public static function getSubscribedCommand()
+    {
+        return Commands::DISPATCH_ASYNC_EVENTS;
+    }
 }

pkg/async-event-dispatcher/Commands.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace Enqueue\AsyncEventDispatcher;
+
+final class Commands
+{
+    const DISPATCH_ASYNC_EVENTS = 'symfony.dispatch_async_events';
+}

pkg/async-event-dispatcher/DependencyInjection/AsyncEventsPass.php

@@ -9,10 +9,7 @@
 
 class AsyncEventsPass implements CompilerPassInterface
 {
-    /**
-     * {@inheritdoc}
-     */
-    public function process(ContainerBuilder $container)
+    public function process(ContainerBuilder $container): void
     {
         if (false == $container->hasDefinition('enqueue.events.async_listener')) {
             return;
@@ -45,8 +42,9 @@ public function process(ContainerBuilder $container)
                     ;
 
                     $container->getDefinition('enqueue.events.async_processor')
-                        ->addTag('enqueue.client.processor', [
-                            'topicName' => 'event.'.$event,
+                        ->addTag('enqueue.processor', [
+                            'topic' => 'event.'.$event,
+                            'client' => 'default',
                         ])
                     ;
 
@@ -78,8 +76,9 @@ public function process(ContainerBuilder $container)
                         ;
 
                         $container->getDefinition('enqueue.events.async_processor')
-                            ->addTag('enqueue.client.processor', [
+                            ->addTag('enqueue.processor', [
                                 'topicName' => 'event.'.$event,
+                                'client' => 'default',
                             ])
                         ;