[bundle] Async events. Add func tests for async listener

Author: makasim

docs/bundle/async_events.md

@@ -34,7 +34,7 @@ To make your listener async you have add `async: true` attribute to the tag `ker
 ```yaml
 # app/config/config.yml
 
-service:
+services:
     acme.foo_listener:
         class: 'AcmeBundle\Listener\FooListener'
         tags:
@@ -50,7 +50,7 @@ You can also add an async listener directly and register a custom message proces
 ```yaml
 # app/config/config.yml
 
-service:
+services:
     acme.async_foo_listener:
         class: 'Enqueue\Bundle\Events\AsyncListener'
         public: false
@@ -189,7 +189,7 @@ and register it:
 ```yaml
 # app/config/config.yml
 
-service:
+services:
     acme.foo_event_transofrmer:
         class: 'AcmeBundle\Listener\FooEventTransformer'
         arguments: ['@doctrine']

pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php

@@ -116,7 +116,7 @@ public function load(array $configs, ContainerBuilder $container)
         if (isset($config['async_events']['enabled'])) {
             $loader->load('events.yml');
 
-            if (isset($config['async_events']['spool_producer'])) {
+            if (false == empty($config['async_events']['spool_producer'])) {
                 $container->getDefinition('enqueue.events.async_listener')
                     ->replaceArgument(0, new Reference('enqueue.client.spool_producer'))
                 ;

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

@@ -8,13 +8,11 @@ services:
 
     enqueue.events.async_listener:
         class: 'Enqueue\Bundle\Events\AsyncListener'
-        public: false
         arguments: ['@enqueue.client.producer', '@enqueue.events.registry']
 
 
     enqueue.events.event_dispatcher:
         class: 'Enqueue\Bundle\Events\ProxyEventDispatcher'
-        public: false
         arguments:
             - '@event_dispatcher'
             - '@enqueue.events.async_listener'

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

@@ -0,0 +1,13 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Functional\App;
+
+class TestAsyncListener
+{
+    public $calls = [];
+
+    public function onEvent()
+    {
+        $this->calls[] = func_get_args();
+    }
+}

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

@@ -35,5 +35,14 @@ enqueue:
     transport:
         default: 'null'
         'null': ~
-    client: ~
+    client:
+        traceable_producer: true
     job: true
+    async_events: true
+
+
+services:
+    test_async_listener:
+        class: 'Enqueue\Bundle\Tests\Functional\App\TestAsyncListener'
+        tags:
+            - { name: 'kernel.event_listener', async: true, event: 'test_async', method: 'onEvent' }

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

@@ -0,0 +1,93 @@
+<?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;
+
+class AsyncListenerTest extends WebTestCase
+{
+    public function setUp()
+    {
+        parent::setUp();
+
+        /** @var AsyncListener $asyncListener */
+        $asyncListener = $this->container->get('enqueue.events.async_listener');
+
+        $asyncListener->resetSyncMode();
+    }
+
+    public function testShouldNotCallRealListenerIfMarkedAsAsync()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $dispatcher->dispatch('test_async', new GenericEvent('aSubject'));
+
+        /** @var TestAsyncListener $listner */
+        $listner = $this->container->get('test_async_listener');
+
+        $this->assertEmpty($listner->calls);
+    }
+
+    public function testShouldSendMessageToExpectedTopicInsteadOfCallingRealListener()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $event = new GenericEvent('theSubject', ['fooArg' => 'fooVal']);
+
+        $dispatcher->dispatch('test_async', $event);
+
+        /** @var TraceableProducer $producer */
+        $producer = $this->container->get('enqueue.producer');
+
+        $traces = $producer->getTopicTraces('event.test_async');
+
+        $this->assertCount(1, $traces);
+
+        $this->assertEquals('event.test_async', $traces[0]['topic']);
+        $this->assertEquals(serialize($event), $traces[0]['body']);
+    }
+
+    public function testShouldSendMessageForEveryDispatchCall()
+    {
+        /** @var EventDispatcherInterface $dispatcher */
+        $dispatcher = $this->container->get('event_dispatcher');
+
+        $dispatcher->dispatch('test_async', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+        $dispatcher->dispatch('test_async', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+        $dispatcher->dispatch('test_async', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+
+        /** @var TraceableProducer $producer */
+        $producer = $this->container->get('enqueue.producer');
+
+        $traces = $producer->getTopicTraces('event.test_async');
+
+        $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', new GenericEvent('theSubject', ['fooArg' => 'fooVal']));
+        });
+
+        $dispatcher->dispatch('foo');
+
+        /** @var TraceableProducer $producer */
+        $producer = $this->container->get('enqueue.producer');
+
+        $traces = $producer->getTopicTraces('event.test_async');
+
+        $this->assertCount(1, $traces);
+    }
+}