Symfony. Async event subscriber.

docs/bundle/async_events.md

@@ -43,6 +43,18 @@ services:
             - { name: 'kernel.event_listener', async: true, event: 'foo', method: 'onEvent' }
 ```
 
+or to `kernel.event_subscriber`:
+
+```yaml
+# app/config/config.yml
+
+services: 
+    test_async_subscriber:
+        class: 'AcmeBundle\Listener\TestAsyncSubscriber'
+        tags:
+            - { name: 'kernel.event_subscriber', async: true }
+```
+
 That's basically it. The rest of the doc describes advanced features. 
 
 ## Advanced Usage.

pkg/enqueue-bundle/Events/DependencyInjection/AsyncEventsPass.php

@@ -5,6 +5,7 @@
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\EventDispatcher\DependencyInjection\RegisterListenersPass;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 
 class AsyncEventsPass implements CompilerPassInterface
 {
@@ -28,26 +29,62 @@ public function process(ContainerBuilder $container)
                     continue;
                 }
 
+                $event = $tagAttribute['event'];
+
                 $service = $container->getDefinition($serviceId);
 
                 $service->clearTag('kernel.event_listener');
                 $service->addTag('enqueue.async_event_listener', $tagAttribute);
 
-                if (false == isset($registeredToEvent[$tagAttribute['event']])) {
+                if (false == isset($registeredToEvent[$event])) {
                     $container->getDefinition('enqueue.events.async_listener')
                         ->addTag('kernel.event_listener', [
-                            'event' => $tagAttribute['event'],
+                            'event' => $event,
                             'method' => 'onEvent',
                         ])
                     ;
 
                     $container->getDefinition('enqueue.events.async_processor')
                         ->addTag('enqueue.client.processor', [
-                            'topicName' => 'event.'.$tagAttribute['event'],
+                            'topicName' => 'event.'.$event,
                         ])
                     ;
 
-                    $registeredToEvent[$tagAttribute['event']] = true;
+                    $registeredToEvent[$event] = true;
+                }
+            }
+        }
+
+        foreach ($container->findTaggedServiceIds('kernel.event_subscriber') as $serviceId => $tagAttributes) {
+            foreach ($tagAttributes as $tagAttribute) {
+                if (false == isset($tagAttribute['async'])) {
+                    continue;
+                }
+
+                $service = $container->getDefinition($serviceId);
+                $service->clearTag('kernel.event_subscriber');
+                $service->addTag('enqueue.async_event_subscriber', $tagAttribute);
+
+                /** @var EventSubscriberInterface $serviceClass */
+                $serviceClass = $service->getClass();
+
+                foreach ($serviceClass::getSubscribedEvents() as $event => $data) {
+                    if (false == isset($registeredToEvent[$event])) {
+                        $container->getDefinition('enqueue.events.async_listener')
+                            ->addTag('kernel.event_listener', [
+                                'event' => $event,
+                                'method' => 'onEvent',
+                            ])
+                        ;
+
+                        $container->getDefinition('enqueue.events.async_processor')
+                            ->addTag('enqueue.client.processor', [
+                                'topicName' => 'event.'.$event,
+                            ])
+                        ;
+
+                        $registeredToEvent[$event] = true;
+                    }
                 }
             }
         }

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

@@ -0,0 +1,20 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Functional\App;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+
+class TestAsyncSubscriber implements EventSubscriberInterface
+{
+    public $calls = [];
+
+    public function onEvent()
+    {
+        $this->calls[] = func_get_args();
+    }
+
+    public static function getSubscribedEvents()
+    {
+        return ['test_async_subscriber' => 'onEvent'];
+    }
+}

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

@@ -47,7 +47,13 @@ services:
         tags:
             - { name: 'kernel.event_listener', async: true, event: 'test_async', method: 'onEvent' }
 
+    test_async_subscriber:
+        class: 'Enqueue\Bundle\Tests\Functional\App\TestAsyncSubscriber'
+        tags:
+            - { name: 'kernel.event_subscriber', async: true }
+
     test_async_event_transformer:
         class: 'Enqueue\Bundle\Tests\Functional\App\TestAsyncEventTransformer'
         tags:
             - {name: 'enqueue.event_transformer', eventName: 'test_async', transformerName: 'test_async' }
+            - {name: 'enqueue.event_transformer', eventName: 'test_async_subscriber', transformerName: 'test_async' }

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

@@ -5,6 +5,7 @@
 use Enqueue\Bundle\Events\AsyncListener;
 use Enqueue\Bundle\Events\AsyncProcessor;
 use Enqueue\Bundle\Tests\Functional\App\TestAsyncListener;
+use Enqueue\Bundle\Tests\Functional\App\TestAsyncSubscriber;
 use Enqueue\Bundle\Tests\Functional\WebTestCase;
 use Enqueue\Null\NullContext;
 use Enqueue\Null\NullMessage;
@@ -86,4 +87,35 @@ public function testShouldCallRealListener()
             $listener->calls[0][2]
         );
     }
+
+    public function testShouldCallRealSubscriber()
+    {
+        /** @var AsyncProcessor $processor */
+        $processor = $this->container->get('enqueue.events.async_processor');
+
+        $message = new NullMessage();
+        $message->setProperty('event_name', 'test_async_subscriber');
+        $message->setProperty('transformer_name', 'test_async');
+        $message->setBody(JSON::encode([
+            'subject' => 'theSubject',
+            'arguments' => ['fooArg' => 'fooVal'],
+        ]));
+
+        $this->assertEquals(PsrProcessor::ACK, $processor->process($message, new NullContext()));
+
+        /** @var TestAsyncSubscriber $subscriber */
+        $subscriber = $this->container->get('test_async_subscriber');
+
+        $this->assertNotEmpty($subscriber->calls);
+
+        $this->assertInstanceOf(GenericEvent::class, $subscriber->calls[0][0]);
+        $this->assertEquals('theSubject', $subscriber->calls[0][0]->getSubject());
+        $this->assertEquals(['fooArg' => 'fooVal'], $subscriber->calls[0][0]->getArguments());
+        $this->assertEquals('test_async_subscriber', $subscriber->calls[0][1]);
+
+        $this->assertSame(
+            $this->container->get('enqueue.events.event_dispatcher'),
+            $subscriber->calls[0][2]
+        );
+    }
 }

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

@@ -0,0 +1,96 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Functional\Events;
+
+use Enqueue\Bundle\Events\AsyncListener;
+use Enqueue\Bundle\Tests\Functional\App\TestAsyncListener;
+use Enqueue\Bundle\Tests\Functional\WebTestCase;
+use Enqueue\Client\TraceableProducer;
+use Symfony\Component\EventDispatcher\Event;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\EventDispatcher\GenericEvent;
+
+/**
+ * @group functional
+ */
+class AsyncSubscriberTest extends WebTestCase
+{
+    public function setUp()
+    {
+        parent::setUp();
+
+        /** @var AsyncListener $asyncListener */
+        $asyncListener = $this->container->get('enqueue.events.async_listener');
+
+        $asyncListener->resetSyncMode();
+    }
+
+    public function testShouldNotCallRealSubscriberIfMarkedAsAsync()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $dispatcher->dispatch('test_async_subscriber', new GenericEvent('aSubject'));
+
+        /** @var TestAsyncListener $listener */
+        $listener = $this->container->get('test_async_subscriber');
+
+        $this->assertEmpty($listener->calls);
+    }
+
+    public function testShouldSendMessageToExpectedTopicInsteadOfCallingRealSubscriber()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $event = new GenericEvent('theSubject', ['fooArg' => 'fooVal']);
+
+        $dispatcher->dispatch('test_async_subscriber', $event);
+
+        /** @var TraceableProducer $producer */
+        $producer = $this->container->get('enqueue.producer');
+
+        $traces = $producer->getTopicTraces('event.test_async_subscriber');
+
+        $this->assertCount(1, $traces);
+
+        $this->assertEquals('event.test_async_subscriber', $traces[0]['topic']);
+        $this->assertEquals('{"subject":"theSubject","arguments":{"fooArg":"fooVal"}}', $traces[0]['body']);
+    }
+
+    public function testShouldSendMessageForEveryDispatchCall()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $dispatcher->dispatch('test_async_subscriber', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+        $dispatcher->dispatch('test_async_subscriber', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+        $dispatcher->dispatch('test_async_subscriber', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+
+        /** @var TraceableProducer $producer */
+        $producer = $this->container->get('enqueue.producer');
+
+        $traces = $producer->getTopicTraces('event.test_async_subscriber');
+
+        $this->assertCount(3, $traces);
+    }
+
+    public function testShouldSendMessageIfDispatchedFromInsideListener()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $dispatcher->addListener('foo', function (Event $event, $eventName, EventDispatcherInterface $dispatcher) {
+            $dispatcher->dispatch('test_async_subscriber', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+        });
+
+        $dispatcher->dispatch('foo');
+
+        /** @var TraceableProducer $producer */
+        $producer = $this->container->get('enqueue.producer');
+
+        $traces = $producer->getTopicTraces('event.test_async_subscriber');
+
+        $this->assertCount(1, $traces);
+    }
+}