Merge pull request #18 from sanjamedic/master

000 - implement logic for high priority tasker queue

Author: gomboc

src/Consts.php

@@ -51,6 +51,8 @@ class Consts
     const ORDER_BY_NAME_CREATED_ON = 'CREATED_ON';
     const ORDER_BY_NAME_PRIORITY   = 'PRIORITY';
 
+    const PARAM_PRIORITY           = 'priority';
+
     public static function getMap()
     {
         return array(

src/Tasker2/Manager.php

@@ -2,6 +2,7 @@
 
 namespace G4\Tasker\Tasker2;
 
+use G4\Tasker\Consts;
 use Model\Domain\RabbitMq\RabbitMqConsts;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
 use PhpAmqpLib\Message\AMQPMessage;
@@ -76,11 +77,18 @@ private function addToMessageQueue()
         try {
             $messages = $this->getMessages();
             foreach ($messages as $message) {
+                $decodedMessageBody = json_decode($message->getBody(), 1);
+                $binding = ($this->messageOptions->hasBindingHP() && isset($decodedMessageBody[Consts::PARAM_PRIORITY])
+                    && ($decodedMessageBody[Consts::PARAM_PRIORITY] > Consts::PRIORITY_50))
+                    ? $this->messageOptions->getBindingHP()
+                    : $this->messageOptions->getBinding();
+
                 $channel->batch_basic_publish(
                     $message,
                     $this->messageOptions->getExchange(),
-                    $this->messageOptions->getBinding()
+                    $binding
                 );
+
             }
             $channel->publish_batch();
         } catch (\Exception $e) {

src/Tasker2/MessageOptions.php

@@ -18,6 +18,11 @@ class MessageOptions
      */
     private $binding;
 
+    /**
+     * @var StringLiteral
+     */
+    private $bindingHP;
+
     /**
      * @var int
      */
@@ -54,4 +59,26 @@ public function getDeliveryMode()
         return (int) $this->deliveryMode;
     }
 
+    /**
+     * @return string
+     */
+    public function getBindingHP()
+    {
+        return (string) $this->bindingHP;
+    }
+
+    /**
+     * @return bool
+     */
+    public function hasBindingHP()
+    {
+        return isset($this->bindingHP) && $this->bindingHP instanceof StringLiteral;
+    }
+
+    public function setBindingHP(StringLiteral $bindingHP)
+    {
+        $this->bindingHP = $bindingHP;
+        return $this;
+    }
+
 }

src/Tasker2/TaskQueue.php

@@ -2,6 +2,7 @@
 
 namespace G4\Tasker\Tasker2;
 
+use G4\Tasker\Consts;
 use G4\Tasker\TaskAbstract;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
 use PhpAmqpLib\Message\AMQPMessage;
@@ -105,10 +106,15 @@ private function saveCurrentTasks($tasks)
         }, $tasks);
 
         foreach ($messages as $message) {
+            $decodedMessageBody = json_decode($message->getBody(), 1);
+            $binding = ($this->messageOptions->hasBindingHP() && isset($decodedMessageBody[Consts::PARAM_PRIORITY])
+                && ($decodedMessageBody[Consts::PARAM_PRIORITY] > Consts::PRIORITY_50))
+                ? $this->messageOptions->getBindingHP()
+                : $this->messageOptions->getBinding();
             $channel->batch_basic_publish(
                 $message,
                 $this->messageOptions->getExchange(),
-                $this->messageOptions->getBinding()
+                $binding
             );
         }
         $channel->publish_batch();