Symfony. Async event dispatching

pkg/enqueue-bundle/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()

pkg/enqueue-bundle/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');
         }

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, $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,52 @@
+<?php
+
+namespace Enqueue\Bundle\Events;
+
+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 self::REJECT;
+        }
+        if (false == $transformerName = $message->getProperty('transformer_name')) {
+            return self::REJECT;
+        }
+
+        $event = $this->registry->getTransformer($transformerName)->toEvent($eventName, $message);
+
+        $this->eventDispatcher->resetSyncMode();
+        $this->eventDispatcher->syncMode($eventName);
+
+        $this->eventDispatcher->dispatch($eventName, $event);
+
+        return self::ACK;
+    }
+}

pkg/enqueue-bundle/Events/ContainerAwareRegistry.php

@@ -0,0 +1,82 @@
+<?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 $eventNamesMap;
+
+    /**
+     * @var string[]
+     */
+    private $transformerIdsMap;
+
+    /**
+     * @param string[] $eventNamesMap
+     * @param string[] $transformerIdsMap
+     */
+    public function __construct(array $eventNamesMap, array $transformerIdsMap)
+    {
+        $this->eventNamesMap = $eventNamesMap;
+        $this->transformerIdsMap = $transformerIdsMap;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getTransformerNameForEvent($eventName)
+    {
+        $transformerName = null;
+        if (array_key_exists($eventName, $this->eventNamesMap)) {
+            $transformerName = $this->eventNamesMap[$eventName];
+        } else {
+            foreach ($this->eventNamesMap as $eventNamePattern => $name) {
+                if ('/' != $eventNamePattern[0]) {
+                    continue;
+                }
+
+                if (preg_match($eventNamePattern, $eventName)) {
+                    $transformerName = $name;
+
+                    break;
+                }
+            }
+        }
+
+        if (empty($transformerName)) {
+            throw new \LogicException(sprintf('There is no transformer registered for the given event %s', $eventName));
+        }
+
+        return $transformerName;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getTransformer($name)
+    {
+        if (false == array_key_exists($name, $this->transformerIdsMap)) {
+            throw new \LogicException(sprintf('There is no transformer named %s', $name));
+        }
+
+        $transformer = $this->container->get($this->transformerIdsMap[$name]);
+
+        if (false == $transformer instanceof  EventTransformer) {
+            throw new \LogicException(sprintf(
+                'The container must return instance of %s but got %s',
+                EventTransformer::class,
+                is_object($transformer) ? get_class($transformer) : gettype($transformer)
+            ));
+        }
+
+        return $transformer;
+    }
+}

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

@@ -0,0 +1,62 @@
+<?php
+
+namespace Enqueue\Bundle\Events\DependencyInjection;
+
+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;
+        }
+
+        if (false == $container->hasDefinition('enqueue.events.registry')) {
+            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',
+                        ])
+                    ;
+
+                    $container->getDefinition('enqueue.events.async_processor')
+                        ->addTag('enqueue.client.processor', [
+                            'topicName' => 'event.'.$tagAttribute['event'],
+                        ])
+                    ;
+
+                    $registeredToEvent[$tagAttribute['event']] = true;
+                }
+            }
+        }
+
+        $registerListenersPass = new RegisterListenersPass(
+            'enqueue.events.event_dispatcher',
+            'enqueue.async_event_listener',
+            'enqueue.async_event_subscriber'
+        );
+        $registerListenersPass->process($container);
+    }
+}

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

@@ -0,0 +1,41 @@
+<?php
+
+namespace Enqueue\Bundle\Events\DependencyInjection;
+
+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;
+        }
+
+        $transformerIdsMap = [];
+        $eventNamesMap = [];
+        foreach ($container->findTaggedServiceIds('enqueue.event_transformer') as $serviceId => $tagAttributes) {
+            foreach ($tagAttributes as $tagAttribute) {
+                if (false == isset($tagAttribute['eventName'])) {
+                    throw new \LogicException('The eventName attribute must be set');
+                }
+
+                $eventName = $tagAttribute['eventName'];
+
+                $transformerName = isset($tagAttribute['transformerName']) ? $tagAttribute['transformerName'] : $serviceId;
+
+                $eventNamesMap[$eventName] = $transformerName;
+                $transformerIdsMap[$transformerName] = $serviceId;
+            }
+        }
+
+        $container->getDefinition('enqueue.events.registry')
+            ->replaceArgument(0, $eventNamesMap)
+            ->replaceArgument(1, $transformerIdsMap)
+        ;
+    }
+}

pkg/enqueue-bundle/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);
+}