[async-event] Move async event classes to own package.

Author: makasim

.gitignore

@@ -4,3 +4,4 @@
 /phpunit.xml
 /vendor/
 /.idea/
+Tests/Functional/queues

ProxyEventDispatcher.php renamed to AsyncEventDispatcher.php

@@ -6,7 +6,7 @@
 use Symfony\Component\EventDispatcher\EventDispatcher;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
-class ProxyEventDispatcher extends EventDispatcher
+class AsyncEventDispatcher extends EventDispatcher
 {
     /**
      * @var EventDispatcherInterface

AsyncListener.php

@@ -2,35 +2,42 @@
 
 namespace Enqueue\AsyncEventDispatcher;
 
-use Enqueue\Client\Message;
-use Enqueue\Client\ProducerInterface;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrQueue;
 use Symfony\Component\EventDispatcher\Event;
 
 class AsyncListener
 {
     /**
-     * @var ProducerInterface
+     * @var PsrContext
      */
-    private $producer;
+    private $context;
 
     /**
      * @var Registry
      */
     private $registry;
 
+    /**
+     * @var PsrQueue
+     */
+    private $eventQueue;
+
     /**
      * @var bool
      */
     private $syncMode;
 
     /**
-     * @param ProducerInterface $producer
-     * @param Registry          $registry
+     * @param PsrContext      $context
+     * @param Registry        $registry
+     * @param PsrQueue|string $eventQueue
      */
-    public function __construct(ProducerInterface $producer, Registry $registry)
+    public function __construct(PsrContext $context, Registry $registry, $eventQueue)
     {
-        $this->producer = $producer;
+        $this->context = $context;
         $this->registry = $registry;
+        $this->eventQueue = $eventQueue instanceof PsrQueue ? $eventQueue : $context->createQueue($eventQueue);
     }
 
     public function resetSyncMode()
@@ -46,6 +53,16 @@ public function syncMode($eventName)
         $this->syncMode[$eventName] = true;
     }
 
+    /**
+     * @param string $eventName
+     *
+     * @return bool
+     */
+    public function isSyncMode($eventName)
+    {
+        return isset($this->syncMode[$eventName]);
+    }
+
     /**
      * @param Event  $event
      * @param string $eventName
@@ -56,11 +73,10 @@ public function onEvent(Event $event = null, $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->sendEvent('event.'.$eventName, $message);
+            $this->context->createProducer()->send($this->eventQueue, $message);
         }
     }
 }

AsyncProcessor.php

@@ -6,6 +6,7 @@
 use Enqueue\Psr\PsrContext;
 use Enqueue\Psr\PsrMessage;
 use Enqueue\Psr\PsrProcessor;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 class AsyncProcessor implements PsrProcessor
 {
@@ -15,18 +16,28 @@ class AsyncProcessor implements PsrProcessor
     private $registry;
 
     /**
-     * @var ProxyEventDispatcher
+     * @var AsyncEventDispatcher|OldAsyncEventDispatcher
      */
-    private $eventDispatcher;
+    private $dispatcher;
 
     /**
-     * @param Registry             $registry
-     * @param ProxyEventDispatcher $eventDispatcher
+     * @param Registry                 $registry
+     * @param EventDispatcherInterface $dispatcher
      */
-    public function __construct(Registry $registry, ProxyEventDispatcher $eventDispatcher)
+    public function __construct(Registry $registry, EventDispatcherInterface $dispatcher)
     {
         $this->registry = $registry;
-        $this->eventDispatcher = $eventDispatcher;
+
+        if (false == ($dispatcher instanceof AsyncEventDispatcher || $dispatcher instanceof OldAsyncEventDispatcher)) {
+            throw new \InvalidArgumentException(sprintf(
+                'The dispatcher argument must be either instance of "%s" or "%s" but got "%s"',
+                AsyncEventDispatcher::class,
+                OldAsyncEventDispatcher::class,
+                get_class($dispatcher)
+            ));
+        }
+
+        $this->dispatcher = $dispatcher;
     }
 
     /**
@@ -43,7 +54,7 @@ public function process(PsrMessage $message, PsrContext $context)
 
         $event = $this->registry->getTransformer($transformerName)->toEvent($eventName, $message);
 
-        $this->eventDispatcher->dispatchAsyncListenersOnly($eventName, $event);
+        $this->dispatcher->dispatchAsyncListenersOnly($eventName, $event);
 
         return self::ACK;
     }

DependencyInjection/AsyncEventDispatcherExtension.php

@@ -2,7 +2,7 @@
 
 namespace Enqueue\AsyncEventDispatcher\DependencyInjection;
 
-use Enqueue\AsyncEventDispatcher\OldProxyEventDispatcher;
+use Enqueue\AsyncEventDispatcher\OldAsyncEventDispatcher;
 use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Definition;
@@ -18,11 +18,15 @@ class AsyncEventDispatcherExtension extends Extension
      */
     public function load(array $configs, ContainerBuilder $container)
     {
+        $config = $this->processConfiguration(new Configuration(), $configs);
+
+        $container->setAlias('enqueue.events.context', $config['context_service']);
+
         $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('services.yml');
 
         if (version_compare(Kernel::VERSION, '3.3', '<')) {
-            $container->setDefinition('enqueue.events.event_dispatcher', new Definition(OldProxyEventDispatcher::class, [
+            $container->setDefinition('enqueue.events.event_dispatcher', new Definition(OldAsyncEventDispatcher::class, [
                 new Reference('service_container'),
                 new Reference('event_dispatcher'),
                 new Reference('enqueue.events.async_listener'),

DependencyInjection/Configuration.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace Enqueue\AsyncEventDispatcher\DependencyInjection;
+
+use Symfony\Component\Config\Definition\Builder\TreeBuilder;
+use Symfony\Component\Config\Definition\ConfigurationInterface;
+
+class Configuration implements ConfigurationInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function getConfigTreeBuilder()
+    {
+        $tb = new TreeBuilder();
+        $rootNode = $tb->root('enqueue_async_event_dispatcher');
+
+        $rootNode->children()
+            ->scalarNode('context_service')->isRequired()->cannotBeEmpty()->end()
+        ;
+
+        return $tb;
+    }
+}

EventTransformer.php

@@ -2,8 +2,6 @@
 
 namespace Enqueue\AsyncEventDispatcher;
 
-use Enqueue\Client\Message;
-use Enqueue\Consumption\Result;
 use Enqueue\Psr\PsrMessage;
 use Symfony\Component\EventDispatcher\Event;
 
@@ -13,18 +11,19 @@ interface EventTransformer
      * @param string     $eventName
      * @param Event|null $event
      *
-     * @return Message
+     * @return PsrMessage
      */
-    public function toMessage($eventName, Event $event = null);
+    public function toMessage($eventName, Event $event);
 
     /**
      * If you able to transform message back to event return it.
-     * If you failed to transform for some reason you can return instance of Result object ( Like this Result::reject() );.
+     * If you failed to transform for some reason you can return a string status (@see PsrProcess constants) or an object that implements __toString method.
+     * The object must have a __toString method is supposed to be used as PsrProcessor::process return value.
      *
      * @param string     $eventName
      * @param PsrMessage $message
      *
-     * @return Event|Result|null
+     * @return Event|string|object
      */
     public function toEvent($eventName, PsrMessage $message);
 }

OldProxyEventDispatcher.php renamed to OldAsyncEventDispatcher.php

@@ -7,7 +7,7 @@
 use Symfony\Component\EventDispatcher\Event;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
-class OldProxyEventDispatcher extends ContainerAwareEventDispatcher
+class OldAsyncEventDispatcher extends ContainerAwareEventDispatcher
 {
     /**
      * @var EventDispatcherInterface

PhpSerializerEventTransformer.php

@@ -2,24 +2,41 @@
 
 namespace Enqueue\AsyncEventDispatcher;
 
-use Enqueue\Client\Message;
+use Enqueue\Psr\PsrContext;
 use Enqueue\Psr\PsrMessage;
 use Symfony\Component\EventDispatcher\Event;
 use Symfony\Component\HttpKernel\Kernel;
 
 class PhpSerializerEventTransformer implements EventTransformer
 {
+    /**
+     * @var PsrContext
+     */
+    private $context;
+
+    /**
+     * @var bool
+     */
+    private $skipSymfonyVersionCheck;
+
+    /**
+     * @param PsrContext $context
+     * @param bool       $skipSymfonyVersionCheck It is useful when async dispatcher is used without Kernel. So there is no way to check the version.
+     */
+    public function __construct(PsrContext $context, $skipSymfonyVersionCheck = false)
+    {
+        $this->context = $context;
+        $this->skipSymfonyVersionCheck = $skipSymfonyVersionCheck;
+    }
+
     /**
      * {@inheritdoc}
      */
     public function toMessage($eventName, Event $event = null)
     {
         $this->assertSymfony30OrHigher();
 
-        $message = new Message();
-        $message->setBody(serialize($event));
-
-        return $message;
+        return $this->context->createMessage(serialize($event));
     }
 
     /**
@@ -34,6 +51,10 @@ public function toEvent($eventName, PsrMessage $message)
 
     private function assertSymfony30OrHigher()
     {
+        if ($this->skipSymfonyVersionCheck) {
+            return;
+        }
+
         if (version_compare(Kernel::VERSION, '3.0', '<')) {
             throw new \LogicException(
                 'This transformer does not work on Symfony prior 3.0. '.

Resources/config/services.yml

@@ -1,4 +1,10 @@
+parameters:
+    enqueue_events_queue: 'symfony_events'
+
 services:
+    # should be defined by the extension
+    # enqueue.events.context:
+
     enqueue.events.registry:
         class: 'Enqueue\AsyncEventDispatcher\ContainerAwareRegistry'
         public: false
@@ -8,11 +14,11 @@ services:
 
     enqueue.events.async_listener:
         class: 'Enqueue\AsyncEventDispatcher\AsyncListener'
-        arguments: ['@enqueue.client.producer', '@enqueue.events.registry']
+        arguments: ['@enqueue.events.context', '@enqueue.events.registry', '%enqueue_events_queue%']
 
 
     enqueue.events.event_dispatcher:
-        class: 'Enqueue\AsyncEventDispatcher\ProxyEventDispatcher'
+        class: 'Enqueue\AsyncEventDispatcher\AsyncEventDispatcher'
         arguments:
             - '@event_dispatcher'
             - '@enqueue.events.async_listener'
@@ -22,8 +28,18 @@ services:
         arguments:
             - '@enqueue.events.registry'
             - '@enqueue.events.event_dispatcher'
+        tags:
+            -
+                name: 'enqueue.client.processor'
+                topicName: '__command__'
+                processorName: '%enqueue_events_queue%'
+                queueName: '%enqueue_events_queue%'
+                queueNameHardcoded: true
+                exclusive: true
 
     enqueue.events.php_serializer_event_transofrmer:
         class: 'Enqueue\AsyncEventDispatcher\PhpSerializerEventTransformer'
+        arguments:
+            - '@enqueue.events.context'
         tags:
-            - {name: 'enqueue.event_transformer', eventName: '/.*/', transformerName: 'php_serializer' }
+            - {name: 'enqueue.event_transformer', eventName: '/.*/', transformerName: 'php_serializer' }