Add events for before and after producer publishes a message

Author: nathanjrobertson

Event/AMQPEvent.php

@@ -3,8 +3,8 @@
 namespace OldSound\RabbitMqBundle\Event;
 
 use OldSound\RabbitMqBundle\RabbitMq\Consumer;
+use OldSound\RabbitMqBundle\RabbitMq\Producer;
 use PhpAmqpLib\Message\AMQPMessage;
-use Symfony\Component\EventDispatcher\Event;
 
 /**
  * Class AMQPEvent
@@ -18,6 +18,8 @@ class AMQPEvent extends AbstractAMQPEvent
     public const ON_IDLE                   = 'on_idle';
     public const BEFORE_PROCESSING_MESSAGE = 'before_processing';
     public const AFTER_PROCESSING_MESSAGE  = 'after_processing';
+    public const BEFORE_PUBLISH_MESSAGE = 'before_publishing';
+    public const AFTER_PUBLISH_MESSAGE  = 'after_publishing';
 
     /**
      * @var AMQPMessage
@@ -29,6 +31,11 @@ class AMQPEvent extends AbstractAMQPEvent
      */
     protected $consumer;
 
+    /**
+     * @var Producer
+     */
+    protected $producer;
+
     /**
      * @return AMQPMessage
      */
@@ -68,4 +75,24 @@ public function setConsumer(Consumer $consumer)
 
         return $this;
     }
+
+    /**
+     * @return Producer
+     */
+    public function getProducer()
+    {
+        return $this->producer;
+    }
+
+    /**
+     * @param Producer $producer
+     *
+     * @return AMQPEvent
+     */
+    public function setProducer(Producer $producer)
+    {
+        $this->producer = $producer;
+
+        return $this;
+    }
 }

Event/AfterProducerPublishMessageEvent.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Event;
+
+use OldSound\RabbitMqBundle\RabbitMq\Producer;
+use PhpAmqpLib\Message\AMQPMessage;
+
+/**
+ * Class AfterProducerPublishMessageEvent
+ *
+ * @package OldSound\RabbitMqBundle\Command
+ */
+class AfterProducerPublishMessageEvent extends AMQPEvent
+{
+    public const NAME = AMQPEvent::AFTER_PROCESSING_MESSAGE;
+
+    /**
+     * @var string
+     */
+    protected $routingKey;
+
+    /**
+     * AfterProducerPublishMessageEvent constructor.
+     *
+     * @param AMQPMessage $AMQPMessage
+     */
+    public function __construct(Producer $producer, AMQPMessage $AMQPMessage, string $routingKey)
+    {
+        $this->setProducer($producer);
+        $this->setAMQPMessage($AMQPMessage);
+        $this->routingKey = $routingKey;
+    }
+
+    /**
+     * @return AMQPMessage
+     */
+    public function getRoutingKey()
+    {
+        return $this->routingKey;
+    }
+}

Event/BeforeProducerPublishMessageEvent.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Event;
+
+use OldSound\RabbitMqBundle\RabbitMq\Producer;
+use PhpAmqpLib\Message\AMQPMessage;
+
+/**
+ * Class BeforeProducerPublishMessageEvent
+ *
+ * @package OldSound\RabbitMqBundle\Command
+ */
+class BeforeProducerPublishMessageEvent extends AMQPEvent
+{
+    public const NAME = AMQPEvent::BEFORE_PROCESSING_MESSAGE;
+
+    /**
+     * @var string
+     */
+    protected $routingKey;
+
+    /**
+     * BeforeProducerPublishMessageEvent constructor.
+     *
+     * @param AMQPMessage $AMQPMessage
+     */
+    public function __construct(Producer $producer, AMQPMessage $AMQPMessage, string $routingKey)
+    {
+        $this->setProducer($producer);
+        $this->setAMQPMessage($AMQPMessage);
+        $this->routingKey = $routingKey;
+    }
+
+    /**
+     * @return AMQPMessage
+     */
+    public function getRoutingKey()
+    {
+        return $this->routingKey;
+    }
+}

RabbitMq/Producer.php

@@ -2,6 +2,8 @@
 
 namespace OldSound\RabbitMqBundle\RabbitMq;
 
+use OldSound\RabbitMqBundle\Event\AfterProducerPublishMessageEvent;
+use OldSound\RabbitMqBundle\Event\BeforeProducerPublishMessageEvent;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
 
@@ -63,6 +65,12 @@ public function publish($msgBody, $routingKey = null, $additionalProperties = []
         }
 
         $real_routingKey = $routingKey !== null ? $routingKey : $this->defaultRoutingKey;
+
+        $this->dispatchEvent(
+            BeforeProducerPublishMessageEvent::NAME,
+            new BeforeProducerPublishMessageEvent($this, $msg, $real_routingKey)
+        );
+
         $this->getChannel()->basic_publish($msg, $this->exchangeOptions['name'], (string)$real_routingKey);
         $this->logger->debug('AMQP message published', [
             'amqp' => [
@@ -72,5 +80,10 @@ public function publish($msgBody, $routingKey = null, $additionalProperties = []
                 'headers' => $headers,
             ],
         ]);
+
+        $this->dispatchEvent(
+            AfterProducerPublishMessageEvent::NAME,
+            new AfterProducerPublishMessageEvent($this, $msg, $real_routingKey)
+        );
     }
 }