Async event dispatcher

Author: makasim

DependencyInjection/Compiler/AsyncEventsPass.php

@@ -0,0 +1,52 @@
+<?php
+
+namespace Enqueue\Bundle\DependencyInjection\Compiler;
+
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\EventDispatcher\DependencyInjection\RegisterListenersPass;
+
+class AsyncEventsPass implements CompilerPassInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function process(ContainerBuilder $container)
+    {
+        if (false == $container->hasDefinition('enqueue.events.async_listener')) {
+            return;
+        }
+
+        $registeredToEvent = [];
+        foreach ($container->findTaggedServiceIds('kernel.event_listener') as $serviceId => $tagAttributes) {
+            foreach ($tagAttributes as $tagAttribute) {
+                if (false == isset($tagAttribute['async'])) {
+                    continue;
+                }
+
+                $service = $container->getDefinition($serviceId);
+
+                $service->clearTag('kernel.event_listener');
+                $service->addTag('enqueue.async_event_listener', $tagAttribute);
+
+                if (false == isset($registeredToEvent[$tagAttribute['event']])) {
+                    $container->getDefinition('enqueue.events.async_listener')
+                        ->addTag('kernel.event_listener', [
+                            'event' => $tagAttribute['event'],
+                            'method' => 'onEvent',
+                        ])
+                    ;
+
+                    $registeredToEvent[$tagAttribute['event']] = true;
+                }
+            }
+        }
+
+        $registerListenersPass = new RegisterListenersPass(
+            'enqueue.events.event_dispatcher',
+            'enqueue.async_event_listener',
+            'enqueue.async_event_subscriber'
+        );
+        $registerListenersPass->process($container);
+    }
+}

DependencyInjection/Compiler/AsyncTransformersPass.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace Enqueue\Bundle\DependencyInjection\Compiler;
+
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+
+class AsyncTransformersPass implements CompilerPassInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function process(ContainerBuilder $container)
+    {
+        if (false == $container->hasDefinition('enqueue.events.registry')) {
+            return;
+        }
+
+        $map = [];
+        foreach ($container->findTaggedServiceIds('enqueue.event_transformer') as $serviceId => $tagAttributes) {
+            foreach ($tagAttributes as $tagAttribute) {
+                $map[$tagAttribute['event']] = $serviceId;
+            }
+        }
+
+        $container->getDefinition('enqueue.events.registry')
+            ->replaceArgument(0, $map)
+        ;
+    }
+}

DependencyInjection/Configuration.php

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

DependencyInjection/EnqueueExtension.php

@@ -112,6 +112,10 @@ public function load(array $configs, ContainerBuilder $container)
             $loader->load('job.yml');
         }
 
+        if ($config['async_events']) {
+            $loader->load('events.yml');
+        }
+
         if ($config['extensions']['doctrine_ping_connection_extension']) {
             $loader->load('extensions/doctrine_ping_connection_extension.yml');
         }

EnqueueBundle.php

@@ -5,6 +5,8 @@
 use Enqueue\AmqpExt\AmqpContext;
 use Enqueue\AmqpExt\Symfony\AmqpTransportFactory;
 use Enqueue\AmqpExt\Symfony\RabbitMqAmqpTransportFactory;
+use Enqueue\Bundle\DependencyInjection\Compiler\AsyncEventsPass;
+use Enqueue\Bundle\DependencyInjection\Compiler\AsyncTransformersPass;
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildClientExtensionsPass;
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildClientRoutingPass;
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildConsumptionExtensionsPass;
@@ -39,6 +41,8 @@ public function build(ContainerBuilder $container)
         $container->addCompilerPass(new BuildTopicMetaSubscribersPass());
         $container->addCompilerPass(new BuildQueueMetaRegistryPass());
         $container->addCompilerPass(new BuildClientExtensionsPass());
+        $container->addCompilerPass(new AsyncEventsPass());
+        $container->addCompilerPass(new AsyncTransformersPass());
 
         /** @var EnqueueExtension $extension */
         $extension = $container->getExtension('enqueue');

Events/AsyncListener.php

@@ -0,0 +1,49 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+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 syncMode($eventName)
+    {
+        $this->syncMode[$eventName] = true;
+    }
+
+    public function onEvent(Event $event, $eventName)
+    {
+        if (false == isset($this->syncMode[$eventName])) {
+            $message = $this->registry->getTransformer($eventName)->toMessage($eventName, $event);
+            $message->setProperty('event_name', $eventName);
+
+            $this->producer->send('symfony_events', $message);
+        }
+    }
+}

Events/AsyncProcessor.php

@@ -0,0 +1,58 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+use Enqueue\Client\TopicSubscriberInterface;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Psr\PsrProcessor;
+
+class AsyncProcessor implements PsrProcessor, TopicSubscriberInterface
+{
+    /**
+     * @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 self::REJECT;
+        }
+
+        // TODO set transformer's name explicitly when sending a message.
+
+        $event = $this->registry->getTransformer($eventName)->toEvent($eventName, $message);
+
+        $this->eventDispatcher->syncMode($eventName);
+        $this->eventDispatcher->dispatch($eventName, $event);
+
+        return self::ACK;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedTopics()
+    {
+        return ['symfony_events'];
+    }
+}

Events/ContainerAwareRegistry.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+use Symfony\Component\DependencyInjection\ContainerAwareInterface;
+use Symfony\Component\DependencyInjection\ContainerAwareTrait;
+
+class ContainerAwareRegistry implements Registry, ContainerAwareInterface
+{
+    use ContainerAwareTrait;
+
+    /**
+     * @var string[]
+     */
+    private $transformersMap;
+
+    /**
+     * @param string[] $transformersMap
+     */
+    public function __construct(array $transformersMap)
+    {
+        $this->transformersMap = $transformersMap;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getTransformer($eventName)
+    {
+        if (false == array_key_exists($eventName, $this->transformersMap)) {
+            throw new \LogicException(sprintf('There is no transformer registered for the given event %s', $eventName));
+        }
+
+        // TODO add check container returns instance of EventTransformer interface.
+
+        return $this->container->get($this->transformersMap[$eventName]);
+    }
+}

Events/EventTransformer.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+use Enqueue\Client\Message;
+use Enqueue\Psr\PsrMessage;
+use Symfony\Component\EventDispatcher\Event;
+
+interface EventTransformer
+{
+    /**
+     * @param string     $eventName
+     * @param Event|null $event
+     *
+     * @return Message
+     */
+    public function toMessage($eventName, Event $event = null);
+
+    /**
+     * @param string     $eventName
+     * @param PsrMessage $message
+     *
+     * @return Event|null
+     */
+    public function toEvent($eventName, PsrMessage $message);
+}

Events/ProxyEventDispatcher.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+use Symfony\Component\EventDispatcher\Event;
+use Symfony\Component\EventDispatcher\EventDispatcher;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+
+class ProxyEventDispatcher extends EventDispatcher
+{
+    /**
+     * @var EventDispatcherInterface
+     */
+    private $trueEventDispatcher;
+
+    /**
+     * @var AsyncListener
+     */
+    private $asyncListener;
+
+    /**
+     * @param EventDispatcherInterface $trueEventDispatcher
+     * @param AsyncListener            $asyncListener
+     */
+    public function __construct(EventDispatcherInterface $trueEventDispatcher, AsyncListener $asyncListener)
+    {
+        $this->trueEventDispatcher = $trueEventDispatcher;
+        $this->asyncListener = $asyncListener;
+    }
+
+    /**
+     * @param string $eventName
+     */
+    public function syncMode($eventName)
+    {
+        $this->asyncListener->syncMode($eventName);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function dispatch($eventName, Event $event = null)
+    {
+        parent::dispatch($eventName, $event);
+        $this->trueEventDispatcher->dispatch($eventName, $event);
+    }
+}