client extension

Author: ASKozienko

pkg/enqueue/Client/ChainExtension.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Enqueue\Client;
+
+class ChainExtension implements ExtensionInterface
+{
+    /**
+     * @var ExtensionInterface[]
+     */
+    private $extensions;
+
+    /**
+     * @param ExtensionInterface[] $extensions
+     */
+    public function __construct(array $extensions)
+    {
+        $this->extensions = $extensions;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function onPreSend($topic, Message $message)
+    {
+        foreach ($this->extensions as $extension) {
+            $extension->onPreSend($topic, $message);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function onPostSend($topic, Message $message)
+    {
+        foreach ($this->extensions as $extension) {
+            $extension->onPostSend($topic, $message);
+        }
+    }
+}

pkg/enqueue/Client/ExtensionInterface.php

@@ -0,0 +1,21 @@
+<?php
+namespace Enqueue\Client;
+
+interface ExtensionInterface
+{
+    /**
+     * @param string  $topic
+     * @param Message $message
+     * @return
+     */
+    public function onPreSend($topic, Message $message);
+
+    /**
+     * @param string $topic
+     * @param Message $message
+     * @return
+     */
+    public function onPostSend($topic, Message $message);
+
+    public function onInterrupted($topic, Message $message);
+}

pkg/enqueue/Client/Producer.php

@@ -12,12 +12,18 @@ class Producer implements ProducerInterface
      */
     protected $driver;
 
+    /**
+     * @var ExtensionInterface
+     */
+    private $extension;
+
     /**
      * @param DriverInterface $driver
      */
-    public function __construct(DriverInterface $driver)
+    public function __construct(DriverInterface $driver, ExtensionInterface $extension = null)
     {
         $this->driver = $driver;
+        $this->extension = $extension ?: new ChainExtension([]);
     }
 
     /**
@@ -47,6 +53,8 @@ public function send($topic, $message)
             $message->setPriority(MessagePriority::NORMAL);
         }
 
+        $this->extension->onPreSend($topic, $message);
+
         if (Message::SCOPE_MESSAGE_BUS == $message->getScope()) {
             if ($message->getProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME)) {
                 throw new \LogicException(sprintf('The %s property must not be set for messages that are sent to message bus.', Config::PARAMETER_PROCESSOR_QUEUE_NAME));
@@ -68,6 +76,8 @@ public function send($topic, $message)
         } else {
             throw new \LogicException(sprintf('The message scope "%s" is not supported.', $message->getScope()));
         }
+
+        $this->extension->onPostSend($topic, $message);
     }
 
     /**