Fix async command\event pkgs

pkg/async-command/DependencyInjection/AsyncCommandExtension.php

@@ -2,28 +2,28 @@
 
 namespace Enqueue\AsyncCommand\DependencyInjection;
 
+use Enqueue\AsyncCommand\Commands;
 use Enqueue\AsyncCommand\RunCommandProcessor;
-use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Extension\Extension;
-use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
 
 class AsyncCommandExtension extends Extension
 {
-    /**
-     * {@inheritdoc}
-     */
     public function load(array $configs, ContainerBuilder $container)
     {
-        $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
-        $loader->load('services.yml');
-
-        $service = $container->register('enqueue.async_command.run_command_processor', RunCommandProcessor::class)
-            ->addArgument('%kernel.project_dir%')
-        ;
-
         foreach ($configs['clients'] as $client) {
-            $service->addTag('enqueue.command_subscriber', ['client' => $client]);
+            $id = sprintf('enqueue.async_command.%s.run_command_processor', $client);
+            $container->register($id, RunCommandProcessor::class)
+                ->addArgument('%kernel.project_dir%')
+                ->addTag('enqueue.processor', [
+                    'client' => $client,
+                    'command' => Commands::RUN_COMMAND,
+                    'queue' => Commands::RUN_COMMAND,
+                    'queue_prefixed' => false,
+                    'exclusive' => true,
+                ])
+                ->addTag('enqueue.transport.processor')
+            ;
         }
     }
 }

pkg/async-command/RunCommandProcessor.php

@@ -2,15 +2,14 @@
 
 namespace Enqueue\AsyncCommand;
 
-use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Consumption\Result;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use Interop\Queue\Processor;
 use Symfony\Component\Process\PhpExecutableFinder;
 use Symfony\Component\Process\Process;
 
-final class RunCommandProcessor implements Processor, CommandSubscriberInterface
+final class RunCommandProcessor implements Processor
 {
     /**
      * @var string
@@ -42,16 +41,6 @@ public function process(Message $message, Context $context): Result
         return Result::ack();
     }
 
-    public static function getSubscribedCommand(): array
-    {
-        return [
-            'command' => Commands::RUN_COMMAND,
-            'queue' => Commands::RUN_COMMAND,
-            'prefix_queue' => false,
-            'exclusive' => true,
-        ];
-    }
-
     /**
      * @return string
      */

pkg/async-command/Tests/RunCommandProcessorTest.php

@@ -2,9 +2,7 @@
 
 namespace Enqueue\AsyncCommand\Tests;
 
-use Enqueue\AsyncCommand\Commands;
 use Enqueue\AsyncCommand\RunCommandProcessor;
-use Enqueue\Client\CommandSubscriberInterface;
 use Interop\Queue\Processor;
 use PHPUnit\Framework\TestCase;
 
@@ -17,13 +15,6 @@ public function testShouldImplementProcessorInterface()
         $this->assertTrue($rc->implementsInterface(Processor::class));
     }
 
-    public function testShouldImplementCommandSubscriberInterfaceInterface()
-    {
-        $rc = new \ReflectionClass(RunCommandProcessor::class);
-
-        $this->assertTrue($rc->implementsInterface(CommandSubscriberInterface::class));
-    }
-
     public function testShouldBeFinal()
     {
         $rc = new \ReflectionClass(RunCommandProcessor::class);
@@ -37,16 +28,4 @@ public function testCouldBeConstructedWithProjectDirAsFirstArgument()
 
         $this->assertAttributeSame('aProjectDir', 'projectDir', $processor);
     }
-
-    public function testShouldSubscribeOnRunCommand()
-    {
-        $subscription = RunCommandProcessor::getSubscribedCommand();
-
-        $this->assertSame([
-            'command' => Commands::RUN_COMMAND,
-            'queue' => Commands::RUN_COMMAND,
-            'prefix_queue' => false,
-            'exclusive' => true,
-        ], $subscription);
-    }
 }

pkg/async-event-dispatcher/AsyncProcessor.php

@@ -2,14 +2,13 @@
 
 namespace Enqueue\AsyncEventDispatcher;
 
-use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Consumption\Result;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use Interop\Queue\Processor;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
-class AsyncProcessor implements Processor, CommandSubscriberInterface
+class AsyncProcessor implements Processor
 {
     /**
      * @var Registry
@@ -51,9 +50,4 @@ public function process(Message $message, Context $context)
 
         return self::ACK;
     }
-
-    public static function getSubscribedCommand()
-    {
-        return Commands::DISPATCH_ASYNC_EVENTS;
-    }
 }

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

@@ -2,17 +2,17 @@
 
 namespace Enqueue\AsyncEventDispatcher\DependencyInjection;
 
+use Enqueue\AsyncEventDispatcher\AsyncProcessor;
+use Enqueue\AsyncEventDispatcher\Commands;
 use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\Alias;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Extension\Extension;
 use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
+use Symfony\Component\DependencyInjection\Reference;
 
 class AsyncEventDispatcherExtension extends Extension
 {
-    /**
-     * {@inheritdoc}
-     */
     public function load(array $configs, ContainerBuilder $container)
     {
         $config = $this->processConfiguration(new Configuration(), $configs);
@@ -21,5 +21,17 @@ public function load(array $configs, ContainerBuilder $container)
 
         $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yml');
+
+        $container->register('enqueue.events.async_processor', AsyncProcessor::class)
+            ->addArgument(new Reference('enqueue.events.registry'))
+            ->addArgument(new Reference('enqueue.events.event_dispatcher'))
+            ->addTag('enqueue.processor', [
+                'command' => Commands::DISPATCH_ASYNC_EVENTS,
+                'queue' => '%enqueue_events_queue%',
+                'queue_prefixed' => false,
+                'exclusive' => true,
+            ])
+            ->addTag('enqueue.transport.processor')
+        ;
     }
 }

pkg/async-event-dispatcher/Resources/config/services.yml

@@ -25,20 +25,6 @@ services:
             - '@event_dispatcher'
             - '@enqueue.events.async_listener'
 
-    enqueue.events.async_processor:
-        class: 'Enqueue\AsyncEventDispatcher\AsyncProcessor'
-        public: public
-        arguments:
-            - '@enqueue.events.registry'
-            - '@enqueue.events.event_dispatcher'
-        tags:
-            -
-                name: 'enqueue.processor'
-                command: 'symfony.dispatch_async_events'
-                queue: '%enqueue_events_queue%'
-                queue_prefixed: false
-                exclusive: true
-
     enqueue.events.php_serializer_event_transofrmer:
         class: 'Enqueue\AsyncEventDispatcher\PhpSerializerEventTransformer'
         public: public

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

@@ -75,6 +75,10 @@ services:
         alias: 'enqueue.client.routes_command'
         public: true
 
+    test.enqueue.events.async_processor:
+        alias: 'enqueue.events.async_processor'
+        public: true
+
     test_async_listener:
         class: 'Enqueue\Bundle\Tests\Functional\App\TestAsyncListener'
         public: true

pkg/enqueue-bundle/Tests/Functional/Events/AsyncProcessorTest.php

@@ -33,15 +33,15 @@ public function setUp()
     public function testCouldBeGetFromContainerAsService()
     {
         /** @var AsyncProcessor $processor */
-        $processor = static::$container->get('enqueue.events.async_processor');
+        $processor = static::$container->get('test.enqueue.events.async_processor');
 
         $this->assertInstanceOf(AsyncProcessor::class, $processor);
     }
 
     public function testShouldRejectIfMessageDoesNotContainEventNameProperty()
     {
         /** @var AsyncProcessor $processor */
-        $processor = static::$container->get('enqueue.events.async_processor');
+        $processor = static::$container->get('test.enqueue.events.async_processor');
 
         $message = new NullMessage();
 
@@ -51,7 +51,7 @@ public function testShouldRejectIfMessageDoesNotContainEventNameProperty()
     public function testShouldRejectIfMessageDoesNotContainTransformerNameProperty()
     {
         /** @var AsyncProcessor $processor */
-        $processor = static::$container->get('enqueue.events.async_processor');
+        $processor = static::$container->get('test.enqueue.events.async_processor');
 
         $message = new NullMessage();
         $message->setProperty('event_name', 'anEventName');
@@ -62,7 +62,7 @@ public function testShouldRejectIfMessageDoesNotContainTransformerNameProperty()
     public function testShouldCallRealListener()
     {
         /** @var AsyncProcessor $processor */
-        $processor = static::$container->get('enqueue.events.async_processor');
+        $processor = static::$container->get('test.enqueue.events.async_processor');
 
         $message = new NullMessage();
         $message->setProperty('event_name', 'test_async');
@@ -93,7 +93,7 @@ public function testShouldCallRealListener()
     public function testShouldCallRealSubscriber()
     {
         /** @var AsyncProcessor $processor */
-        $processor = static::$container->get('enqueue.events.async_processor');
+        $processor = static::$container->get('test.enqueue.events.async_processor');
 
         $message = new NullMessage();
         $message->setProperty('event_name', 'test_async_subscriber');

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

@@ -25,12 +25,12 @@ public function testShouldDisplayRegisteredTopics()
         $tester = new CommandTester($command);
         $tester->execute([]);
 
-        $expected = <<<'OUTPUT'
-| topic   | theTopic                                            | default (prefixed)                     | test_topic_subscriber_processor                         | (hidden) |
-OUTPUT;
-
         $this->assertSame(0, $tester->getStatusCode());
-        $this->assertContains($expected, $tester->getDisplay());
+        $this->assertContains('| topic', $tester->getDisplay());
+        $this->assertContains('| theTopic', $tester->getDisplay());
+        $this->assertContains('| default (prefixed)', $tester->getDisplay());
+        $this->assertContains('| test_topic_subscriber_processor', $tester->getDisplay());
+        $this->assertContains('| (hidden)', $tester->getDisplay());
     }
 
     public function testShouldDisplayCommands()
@@ -41,11 +41,11 @@ public function testShouldDisplayCommands()
         $tester = new CommandTester($command);
         $tester->execute([]);
 
-        $expected = <<<'OUTPUT'
-| command | theCommand                                          | default (prefixed)                     | test_command_subscriber_processor                       | (hidden) |
-OUTPUT;
-
         $this->assertSame(0, $tester->getStatusCode());
-        $this->assertContains($expected, $tester->getDisplay());
+        $this->assertContains('| command', $tester->getDisplay());
+        $this->assertContains('| theCommand', $tester->getDisplay());
+        $this->assertContains('| test_command_subscriber_processor', $tester->getDisplay());
+        $this->assertContains('| default (prefixed)', $tester->getDisplay());
+        $this->assertContains('| (hidden)', $tester->getDisplay());
     }
 }