Merge pull request #86 from php-enqueue/async-events

Symfony. Async event dispatching

Author: makasim

docs/bundle/async_events.md

@@ -0,0 +1,205 @@
+# Async events
+
+The EnqueueBundle allows you to dispatch events asynchronously. 
+Behind the scene it replaces your listener with one that sends a message to MQ. 
+The message contains the event object. 
+The consumer, once it receives the message, restores the event and dispatches it to only async listeners.
+
+Async listeners benefits:
+
+* The response time lesser. It has to do less work.
+* Better fault tolerance. Bugs in async listener does not affect user. Messages will wait till you fix bugs.
+* Better scaling. Add more consumers to meet the load.
+
+_**Note**: The php serializer transformer (the default one) does not work on Symfony prior 3.0. The event contains eventDispatcher and therefor could not be serialized. You have to register a transformer for every async event. Read the [event transformer](#event-transformer)._
+
+## Configuration
+
+I suppose you already [installed the bundle](quick_tour.md#install). 
+Now, you have to enable `async_events`. 
+If you do not enable it, events will be processed as before: synchronously.
+
+```yaml
+# app/config/config.yml
+
+enqueue:
+   async_events:
+      enabled: true
+      # if you'd like to send send messages onTerminate use spool_producer (it makes response time even lesser):
+      # spool_producer: true
+```
+
+## Usage
+
+To make your listener async you have add `async: true` attribute to the tag `kernel.event_listener`, like this:
+
+```yaml
+# app/config/config.yml
+
+services:
+    acme.foo_listener:
+        class: 'AcmeBundle\Listener\FooListener'
+        tags:
+            - { name: 'kernel.event_listener', async: true, event: 'foo', method: 'onEvent' }
+```
+
+That's basically it. The rest of the doc describes advanced features. 
+
+## Advanced Usage.
+
+You can also add an async listener directly and register a custom message processor for it:
+
+```yaml
+# app/config/config.yml
+
+services:
+    acme.async_foo_listener:
+        class: 'Enqueue\Bundle\Events\AsyncListener'
+        public: false
+        arguments: ['@enqueue.client.producer', '@enqueue.events.registry']
+        tags:
+          - { name: 'kernel.event_listener', event: 'foo', method: 'onEvent' }
+```
+
+The message processor must subscribe to `event.foo` topic. The message queue topics names for event follow this patter `event.{eventName}`.
+
+```php
+<?php
+
+use Enqueue\Bundle\Events\Registry;
+use Enqueue\Client\TopicSubscriberInterface;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Psr\PsrProcessor;
+
+class FooEventProcessor implements PsrProcessor, TopicSubscriberInterface
+{
+    /**
+     * @var Registry
+     */
+    private $registry;
+
+    /**
+     * @param Registry $registry
+     */
+    public function __construct(Registry $registry)
+    {
+        $this->registry = $registry;
+    }
+
+    public function process(PsrMessage $message, PsrContext $context)
+    {
+        if (false == $eventName = $message->getProperty('event_name')) {
+            return self::REJECT;
+        }
+        if (false == $transformerName = $message->getProperty('transformer_name')) {
+            return self::REJECT;
+        }
+
+        // do what you want with the event.
+        $event = $this->registry->getTransformer($transformerName)->toEvent($eventName, $message);
+        
+        
+        return self::ACK;
+    }
+
+    public static function getSubscribedTopics()
+    {
+        return ['event.foo'];
+    }
+}
+```
+
+
+## Event transformer
+
+The bundle uses [php serializer](https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/enqueue-bundle/Events/PhpSerializerEventTransformer.php) transformer by default to pass events through MQ.
+You could create a transformer for the given event type. The transformer must implement `Enqueue\Bundle\Events\EventTransformer` interface.
+Consider the next example. It shows how to send an event that contains Doctrine entity as a subject  
+ 
+```php
+<?php
+namespace AcmeBundle\Listener;
+
+// src/AcmeBundle/Listener/FooEventTransformer.php
+
+use Enqueue\Client\Message;
+use Enqueue\Consumption\Result;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Util\JSON;
+use Symfony\Component\EventDispatcher\Event;
+use Enqueue\Bundle\Events\EventTransformer;
+use Doctrine\Bundle\DoctrineBundle\Registry;
+use Symfony\Component\EventDispatcher\GenericEvent;
+
+class FooEventTransformer implements EventTransformer
+{
+    /** @var Registry @doctrine */
+    private $doctrine;
+
+    public function __construct(Registry $doctrine)
+    {
+        $this->doctrine = $doctrine;
+    }
+
+    /**
+     * {@inheritdoc}
+     * 
+     * @param GenericEvent $event
+     */
+    public function toMessage($eventName, Event $event = null)
+    {
+        $entity = $event->getSubject();
+        $entityClass = get_class($event);
+        
+        $manager = $this->doctrine->getManagerForClass($entityClass);
+        $meta = $manager->getClassMetadata($entityClass);
+
+        $id = $meta->getIdentifierValues($entity);
+        
+        $message = new Message();
+        $message->setBody([
+            'entityClass' => $entityClass, 
+            'entityId' => $id,
+            'arguments' => $event->getArguments()
+        ]);
+
+        return $message;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function toEvent($eventName, PsrMessage $message)
+    {
+        $data = JSON::decode($message->getBody());
+        
+        $entityClass = $data['entityClass'];
+        
+        $manager = $this->doctrine->getManagerForClass($entityClass);
+        if (false == $entity = $manager->find($entityClass, $data['entityId'])) {
+            return Result::reject('The entity could not be found.');
+        }
+        
+        return new GenericEvent($entity, $data['arguments']);
+    }
+}
+```
+
+and register it:
+
+```yaml
+# app/config/config.yml
+
+services:
+    acme.foo_event_transofrmer:
+        class: 'AcmeBundle\Listener\FooEventTransformer'
+        arguments: ['@doctrine']
+        tags:
+            - {name: 'enqueue.event_transformer', eventName: 'foo' }
+```
+
+The `eventName` attribute accepts a regexp. You can do next `eventName: '/foo\..*?/'`. 
+It uses this transformer for all event with the name beginning with `foo.`
+
+[back to index](../index.md)

docs/bundle/quick_tour.md

@@ -9,6 +9,8 @@ It adds easy to use [configuration layer](config_reference.md), register service
 $ composer require enqueue/enqueue-bundle enqueue/amqp-ext
 ```
 
+_**Note**: You could use not only AMQP transport but other available: STOMP, Amazon SQS, Redis, Filesystem, Doctrine DBAL and others._
+
 ## Enable the Bundle
 
 Then, enable the bundle by adding `new Enqueue\Bundle\EnqueueBundle()` to the bundles array of the registerBundles method in your project's `app/AppKernel.php` file:

docs/index.md

@@ -27,6 +27,7 @@
     - [Cli commands](bundle/cli_commands.md)
     - [Message producer](bundle/message_producer.md)
     - [Message processor](bundle/message_processor.md)
+    - [Async events](bundle/async_events.md)
     - [Job queue](bundle/job_queue.md)
     - [Consumption extension](bundle/consumption_extension.md)
     - [Production settings](bundle/production_settings.md)

pkg/enqueue-bundle/DependencyInjection/Configuration.php

@@ -51,6 +51,12 @@ public function getConfigTreeBuilder()
                 ->integerNode('redelivered_delay_time')->min(0)->defaultValue(0)->end()
             ->end()->end()
             ->booleanNode('job')->defaultFalse()->end()
+            ->arrayNode('async_events')
+                ->canBeEnabled()
+                ->children()
+                    ->booleanNode('spool_producer')->defaultFalse()->end()
+                ->end()
+            ->end()
             ->arrayNode('extensions')->addDefaultsIfNotSet()->children()
                 ->booleanNode('doctrine_ping_connection_extension')->defaultFalse()->end()
                 ->booleanNode('doctrine_clear_identity_map_extension')->defaultFalse()->end()

pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php

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

pkg/enqueue-bundle/EnqueueBundle.php

@@ -12,6 +12,8 @@
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildQueueMetaRegistryPass;
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildTopicMetaSubscribersPass;
 use Enqueue\Bundle\DependencyInjection\EnqueueExtension;
+use Enqueue\Bundle\Events\DependencyInjection\AsyncEventsPass;
+use Enqueue\Bundle\Events\DependencyInjection\AsyncTransformersPass;
 use Enqueue\Dbal\DbalContext;
 use Enqueue\Dbal\Symfony\DbalTransportFactory;
 use Enqueue\Fs\FsContext;
@@ -23,6 +25,7 @@
 use Enqueue\Stomp\StompContext;
 use Enqueue\Stomp\Symfony\RabbitMqStompTransportFactory;
 use Enqueue\Stomp\Symfony\StompTransportFactory;
+use Symfony\Component\DependencyInjection\Compiler\PassConfig;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\HttpKernel\Bundle\Bundle;
 
@@ -68,5 +71,8 @@ public function build(ContainerBuilder $container)
         if (class_exists(SqsContext::class)) {
             $extension->addTransportFactory(new SqsTransportFactory());
         }
+
+        $container->addCompilerPass(new AsyncEventsPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
+        $container->addCompilerPass(new AsyncTransformersPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
     }
 }

pkg/enqueue-bundle/Events/AsyncListener.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+use Enqueue\Client\Message;
+use Enqueue\Client\ProducerInterface;
+use Symfony\Component\EventDispatcher\Event;
+
+class AsyncListener
+{
+    /**
+     * @var ProducerInterface
+     */
+    private $producer;
+
+    /**
+     * @var Registry
+     */
+    private $registry;
+
+    /**
+     * @var bool
+     */
+    private $syncMode;
+
+    /**
+     * @param ProducerInterface $producer
+     * @param Registry          $registry
+     */
+    public function __construct(ProducerInterface $producer, Registry $registry)
+    {
+        $this->producer = $producer;
+        $this->registry = $registry;
+    }
+
+    public function resetSyncMode()
+    {
+        $this->syncMode = [];
+    }
+
+    /**
+     * @param string $eventName
+     */
+    public function syncMode($eventName)
+    {
+        $this->syncMode[$eventName] = true;
+    }
+
+    /**
+     * @param Event  $event
+     * @param string $eventName
+     */
+    public function onEvent(Event $event = null, $eventName)
+    {
+        if (false == isset($this->syncMode[$eventName])) {
+            $transformerName = $this->registry->getTransformerNameForEvent($eventName);
+
+            $message = $this->registry->getTransformer($transformerName)->toMessage($eventName, $event);
+            $message->setScope(Message::SCOPE_APP);
+            $message->setProperty('event_name', $eventName);
+            $message->setProperty('transformer_name', $transformerName);
+
+            $this->producer->send('event.'.$eventName, $message);
+        }
+    }
+}

pkg/enqueue-bundle/Events/AsyncProcessor.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+use Enqueue\Consumption\Result;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Psr\PsrProcessor;
+
+class AsyncProcessor implements PsrProcessor
+{
+    /**
+     * @var Registry
+     */
+    private $registry;
+
+    /**
+     * @var ProxyEventDispatcher
+     */
+    private $eventDispatcher;
+
+    /**
+     * @param Registry             $registry
+     * @param ProxyEventDispatcher $eventDispatcher
+     */
+    public function __construct(Registry $registry, ProxyEventDispatcher $eventDispatcher)
+    {
+        $this->registry = $registry;
+        $this->eventDispatcher = $eventDispatcher;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function process(PsrMessage $message, PsrContext $context)
+    {
+        if (false == $eventName = $message->getProperty('event_name')) {
+            return Result::reject('The message is missing "event_name" property');
+        }
+        if (false == $transformerName = $message->getProperty('transformer_name')) {
+            return Result::reject('The message is missing "transformer_name" property');
+        }
+
+        $event = $this->registry->getTransformer($transformerName)->toEvent($eventName, $message);
+
+        $this->eventDispatcher->dispatchAsyncListenersOnly($eventName, $event);
+
+        return self::ACK;
+    }
+}