add events

Author: Alexandru Panturu

DependencyInjection/Compiler/InjectEventDispatcherPass.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\DependencyInjection\Compiler;
+
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Reference;
+
+/**
+ * Class InjectEventDispatcherPass
+ *
+ * @package OldSound\RabbitMqBundle\DependencyInjection\Compiler
+ */
+class InjectEventDispatcherPass implements CompilerPassInterface
+{
+    const EVENT_DISPATCHER_SERVICE_ID = 'event_dispatcher';
+
+    /**
+     * @inheritDoc
+     */
+    public function process(ContainerBuilder $container)
+    {
+        if (!$container->has(self::EVENT_DISPATCHER_SERVICE_ID)) {
+            return;
+        }
+        $eventDispatcherDefinition = $container->findDefinition(self::EVENT_DISPATCHER_SERVICE_ID);
+        $taggedConsumers = $container->findTaggedServiceIds('old_sound_rabbit_mq.base_amqp');
+
+        foreach ($taggedConsumers as $id => $tag) {
+            $definition = $container->getDefinition($id);
+            $definition->addMethodCall(
+                'setEventDispatcher',
+                [$eventDispatcherDefinition]
+            );
+        }
+
+    }
+}

Event/AMQPEvent.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Event;
+
+use PhpAmqpLib\Message\AMQPMessage;
+use Symfony\Component\EventDispatcher\Event;
+
+/**
+ * Class AMQPEvent
+ *
+ * @package OldSound\RabbitMqBundle\Event
+ */
+class AMQPEvent extends Event
+{
+    const ON_CONSUME                = 'on_consume';
+    const BEFORE_PROCESSING_MESSAGE = 'before_processing';
+    const AFTER_PROCESSING_MESSAGE  = 'after_processing';
+
+    /**
+     * @var AMQPMessage
+     */
+    protected $AMQPMessage;
+    
+    /**
+     * @return AMQPMessage
+     */
+    public function getAMQPMessage()
+    {
+        return $this->AMQPMessage;
+    }
+}

Event/AfterProcessingMessageEvent.php

@@ -0,0 +1,25 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Event;
+
+use PhpAmqpLib\Message\AMQPMessage;
+
+/**
+ * Class AfterProcessingMessageEvent
+ *
+ * @package OldSound\RabbitMqBundle\Event
+ */
+class AfterProcessingMessageEvent extends AMQPEvent
+{
+    const NAME = AMQPEvent::AFTER_PROCESSING_MESSAGE;
+
+    /**
+     * AfterProcessingMessageEvent constructor.
+     *
+     * @param AMQPMessage $AMQPMessage
+     */
+    public function __construct(AMQPMessage $AMQPMessage)
+    {
+        $this->AMQPMessage = $AMQPMessage;
+    }
+}

Event/BeforeProcessingMessageEvent.php

@@ -0,0 +1,25 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Event;
+
+use PhpAmqpLib\Message\AMQPMessage;
+
+/**
+ * Class BeforeProcessingMessageEvent
+ *
+ * @package OldSound\RabbitMqBundle\Command
+ */
+class BeforeProcessingMessageEvent extends AMQPEvent
+{
+    const NAME = AMQPEvent::BEFORE_PROCESSING_MESSAGE;
+
+    /**
+     * BeforeProcessingMessageEvent constructor.
+     *
+     * @param AMQPMessage $AMQPMessage
+     */
+    public function __construct(AMQPMessage $AMQPMessage)
+    {
+        $this->AMQPMessage = $AMQPMessage;
+    }
+}

Event/OnConsumeEvent.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Event;
+
+/**
+ * Class OnConsumeEvent
+ *
+ * @package OldSound\RabbitMqBundle\Command
+ */
+class OnConsumeEvent extends AMQPEvent
+{
+    const NAME = AMQPEvent::ON_CONSUME;
+}

OldSoundRabbitMqBundle.php

@@ -2,6 +2,7 @@
 
 namespace OldSound\RabbitMqBundle;
 
+use OldSound\RabbitMqBundle\DependencyInjection\Compiler\InjectEventDispatcherPass;
 use OldSound\RabbitMqBundle\DependencyInjection\Compiler\RegisterPartsPass;
 use Symfony\Component\HttpKernel\Bundle\Bundle;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -13,5 +14,6 @@ public function build(ContainerBuilder $container)
         parent::build($container);
 
         $container->addCompilerPass(new RegisterPartsPass());
+        $container->addCompilerPass(new InjectEventDispatcherPass());
     }
 }

RabbitMq/BaseAmqp.php

@@ -6,6 +6,8 @@
 use PhpAmqpLib\Connection\AMQPLazyConnection;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
+use Symfony\Component\EventDispatcher\Event;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 abstract class BaseAmqp
 {
@@ -22,7 +24,7 @@ abstract class BaseAmqp
      * @var LoggerInterface
      */
     protected $logger;
-    
+
     protected $exchangeOptions = array(
         'passive' => false,
         'durable' => true,
@@ -46,6 +48,11 @@ abstract class BaseAmqp
         'declare' => true,
     );
 
+    /**
+     * @var EventDispatcherInterface
+     */
+    protected $eventDispatcher;
+
     /**
      * @param AbstractConnection   $conn
      * @param AMQPChannel|null $ch
@@ -234,4 +241,30 @@ public function setLogger($logger)
     {
         $this->logger = $logger;
     }
+
+    /**
+     * @param EventDispatcherInterface $eventDispatcher
+     *
+     * @return BaseAmqp
+     */
+    public function setEventDispatcher(EventDispatcherInterface $eventDispatcher)
+    {
+        $this->eventDispatcher = $eventDispatcher;
+
+        return $this;
+    }
+
+    /**
+     * @param string $eventName
+     * @param Event  $event
+     */
+    protected function dispatchEvent($eventName, Event $event)
+    {
+        if ($this->eventDispatcher) {
+            $this->eventDispatcher->dispatch(
+                $eventName,
+                $event
+            );
+        }
+    }
 }

RabbitMq/Consumer.php

@@ -2,6 +2,9 @@
 
 namespace OldSound\RabbitMqBundle\RabbitMq;
 
+use OldSound\RabbitMqBundle\Event\AfterProcessingMessageEvent;
+use OldSound\RabbitMqBundle\Event\BeforeProcessingMessageEvent;
+use OldSound\RabbitMqBundle\Event\OnConsumeEvent;
 use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
 use PhpAmqpLib\Message\AMQPMessage;
 
@@ -44,6 +47,7 @@ public function consume($msgAmount)
         $this->setupConsumer();
 
         while (count($this->getChannel()->callbacks)) {
+            $this->dispatchEvent(OnConsumeEvent::NAME, new OnConsumeEvent());
             $this->maybeStopConsumer();
             $this->getChannel()->wait(null, false, $this->getIdleTimeout());
         }
@@ -67,9 +71,16 @@ public function delete()
 
     public function processMessage(AMQPMessage $msg)
     {
+        $this->dispatchEvent(BeforeProcessingMessageEvent::NAME,
+            new BeforeProcessingMessageEvent($msg)
+        );
         try {
             $processFlag = call_user_func($this->callback, $msg);
             $this->handleProcessMessage($msg, $processFlag);
+            $this->dispatchEvent(
+                AfterProcessingMessageEvent::NAME,
+                new AfterProcessingMessageEvent($msg)
+            );
             $this->logger->debug('Queue message processed', array(
                 'amqp' => array(
                     'queue' => $this->queueOptions['name'],