From 8d0c38be091a7f78de5652aad1d1a1d9d89e2d44 Mon Sep 17 00:00:00 2001 From: Maksim Kotlyar Date: Mon, 22 May 2017 16:34:29 +0300 Subject: [PATCH 1/3] Symfony. Async event subscriber. --- .../DependencyInjection/AsyncEventsPass.php | 45 ++++++++- .../Functional/App/TestAsyncSubscriber.php | 20 ++++ .../Tests/Functional/App/config/config.yml | 6 ++ .../Functional/Events/AsyncSubscriberTest.php | 96 +++++++++++++++++++ 4 files changed, 163 insertions(+), 4 deletions(-) create mode 100644 pkg/enqueue-bundle/Tests/Functional/App/TestAsyncSubscriber.php create mode 100644 pkg/enqueue-bundle/Tests/Functional/Events/AsyncSubscriberTest.php diff --git a/pkg/enqueue-bundle/Events/DependencyInjection/AsyncEventsPass.php b/pkg/enqueue-bundle/Events/DependencyInjection/AsyncEventsPass.php index 8d11fe427..f02c041c6 100644 --- a/pkg/enqueue-bundle/Events/DependencyInjection/AsyncEventsPass.php +++ b/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; + } } } } diff --git a/pkg/enqueue-bundle/Tests/Functional/App/TestAsyncSubscriber.php b/pkg/enqueue-bundle/Tests/Functional/App/TestAsyncSubscriber.php new file mode 100644 index 000000000..880519b51 --- /dev/null +++ b/pkg/enqueue-bundle/Tests/Functional/App/TestAsyncSubscriber.php @@ -0,0 +1,20 @@ +calls[] = func_get_args(); + } + + public static function getSubscribedEvents() + { + return ['test_async_subscriber' => 'onEvent']; + } +} diff --git a/pkg/enqueue-bundle/Tests/Functional/App/config/config.yml b/pkg/enqueue-bundle/Tests/Functional/App/config/config.yml index 817c37c7f..ff11521f1 100644 --- a/pkg/enqueue-bundle/Tests/Functional/App/config/config.yml +++ b/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' } diff --git a/pkg/enqueue-bundle/Tests/Functional/Events/AsyncSubscriberTest.php b/pkg/enqueue-bundle/Tests/Functional/Events/AsyncSubscriberTest.php new file mode 100644 index 000000000..e7dd3d8cb --- /dev/null +++ b/pkg/enqueue-bundle/Tests/Functional/Events/AsyncSubscriberTest.php @@ -0,0 +1,96 @@ +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); + } +} From d10e6fb754d37562f1b2d7cb0f96b1614184bbce Mon Sep 17 00:00:00 2001 From: Maksim Kotlyar Date: Mon, 22 May 2017 16:37:57 +0300 Subject: [PATCH 2/3] add subscriber test for async processor --- .../Functional/Events/AsyncProcessorTest.php | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/pkg/enqueue-bundle/Tests/Functional/Events/AsyncProcessorTest.php b/pkg/enqueue-bundle/Tests/Functional/Events/AsyncProcessorTest.php index 549f2a369..fc95843ec 100644 --- a/pkg/enqueue-bundle/Tests/Functional/Events/AsyncProcessorTest.php +++ b/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] + ); + } } From b8a3ce4dde60608b67ddf3c81b84d13f313beb84 Mon Sep 17 00:00:00 2001 From: Maksim Kotlyar Date: Mon, 22 May 2017 16:42:07 +0300 Subject: [PATCH 3/3] [doc] upd async symfony events doc. --- docs/bundle/async_events.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/docs/bundle/async_events.md b/docs/bundle/async_events.md index 00fca9b64..89787c73b 100644 --- a/docs/bundle/async_events.md +++ b/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.