Merge pull request #20 from ppavlovic/master

When RabbitMQ connection is not available add tasks to database

Author: ivankoni

composer.json

@@ -26,7 +26,9 @@
         }
     ],
     "require": {
-        "php"            : ">=5.5",
+        "ext-json"       : "*",
+        "ext-pdo"        : "*",
+        "php"            : ">=5.6",
         "g4/cron"        : "*",
         "g4/log"         : "1.*"
     },
@@ -37,4 +39,4 @@
     "autoload": {
         "psr-4": {"G4\\Tasker\\": "src/"}
     }
-}
+}

src/Tasker2/Manager.php

@@ -2,10 +2,8 @@
 
 namespace G4\Tasker\Tasker2;
 
-use G4\Tasker\Consts;
-use Model\Domain\RabbitMq\RabbitMqConsts;
+use G4\Tasker\Tasker2\Queue\BatchPublisher;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
-use PhpAmqpLib\Message\AMQPMessage;
 
 class Manager
 {
@@ -17,7 +15,7 @@ class Manager
     private $tasks;
 
     /**
-     * @var AMQPStreamConnection
+     * @var AMQPStreamConnection | null
      */
     private $rabbitMqConnection;
 
@@ -35,7 +33,7 @@ class Manager
 
     public function __construct(
         \G4\Tasker\Model\Repository\TaskRepositoryInterface $taskRepository,
-        AMQPStreamConnection $rabbitMqConnection,
+        AMQPStreamConnection $rabbitMqConnection = null,
         MessageOptions $messageOptions
     ) {
         $this->taskRepository = $taskRepository;
@@ -46,6 +44,11 @@ public function __construct(
 
     public function run()
     {
+        if ($this->rabbitMqConnection === null) {
+            // no rabbitmq connection is available
+            trigger_error('RabbitMQ connection is not available for Tasker Manager', E_USER_NOTICE);
+            return;
+        }
         $this
             ->getReservedTasks()
             ->addToMessageQueue()
@@ -85,21 +88,8 @@ 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(),
-                    $binding
-                );
-
-            }
-            $channel->publish_batch();
+            $queuePublisher = new BatchPublisher($channel, $this->messageOptions);
+            $queuePublisher->publish(...$messages);
         } catch (\Exception $e) {
             // todo throw exception if unable to add messages to queue
         } finally {

src/Tasker2/Queue/BatchPublisher.php

@@ -0,0 +1,43 @@
+<?php
+
+namespace G4\Tasker\Tasker2\Queue;
+
+use G4\Tasker\Consts;
+use G4\Tasker\Tasker2\MessageOptions;
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Message\AMQPMessage;
+
+class BatchPublisher
+{
+    /**
+     * @var MessageOptions
+     */
+    private $messageOptions;
+    /**
+     * @var AMQPChannel
+     */
+    private $channel;
+
+    public function __construct(AMQPChannel $channel, MessageOptions $messageOptions)
+    {
+        $this->channel = $channel;
+        $this->messageOptions = $messageOptions;
+    }
+
+    public function publish(AMQPMessage ...$messages)
+    {
+        foreach ($messages as $message) {
+            $decodedMessageBody = json_decode($message->getBody(), true);
+            $binding = ($this->messageOptions->hasBindingHP() && isset($decodedMessageBody[Consts::PARAM_PRIORITY])
+                && ($decodedMessageBody[Consts::PARAM_PRIORITY] > Consts::PRIORITY_50))
+                ? $this->messageOptions->getBindingHP()
+                : $this->messageOptions->getBinding();
+            $this->channel->batch_basic_publish(
+                $message,
+                $this->messageOptions->getExchange(),
+                $binding
+            );
+        }
+        $this->channel->publish_batch();
+    }
+}

src/Tasker2/TaskQueue.php

@@ -2,10 +2,9 @@
 
 namespace G4\Tasker\Tasker2;
 
-use G4\Tasker\Consts;
 use G4\Tasker\TaskAbstract;
+use G4\Tasker\Tasker2\Queue\BatchPublisher;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
-use PhpAmqpLib\Message\AMQPMessage;
 use G4\ValueObject\Uuid;
 
 class TaskQueue
@@ -16,7 +15,7 @@ class TaskQueue
     private $queue;
 
     /**
-     * @var AMQPStreamConnection
+     * @var AMQPStreamConnection | null
      */
     private $AMQPConnection;
 
@@ -37,7 +36,7 @@ class TaskQueue
 
     public function __construct(
         \G4\Tasker\Queue $queue,
-        AMQPStreamConnection $AMQPConnection,
+        AMQPStreamConnection $AMQPConnection = null,
         MessageOptions $messageOptions,
         $requestUuid = null
     )
@@ -97,6 +96,14 @@ private function saveCurrentTasks($tasks)
         if (count($tasks) === 0) {
             return $this;
         }
+
+        if ($this->AMQPConnection === null) {
+            // in case that rabbitmq is not available save tasks to database
+            $this->saveDelayedTasks($tasks);
+            trigger_error('RabbitMQ connection is not available for Tasker TaskQueue', E_USER_NOTICE);
+            return $this;
+        }
+
         $channel = $this->AMQPConnection->channel();
 
         $messages = array_map(function (TaskAbstract $taskAbstract) {
@@ -105,20 +112,10 @@ private function saveCurrentTasks($tasks)
             return (new AmqpMessageFactory($task, $this->messageOptions->getDeliveryMode()))->create();
         }, $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(),
-                $binding
-            );
-        }
-        $channel->publish_batch();
+        $queuePublisher = new BatchPublisher($channel, $this->messageOptions);
+        $queuePublisher->publish(...$messages);
         $channel->close();
+        return $this;
     }
 
     private function getRequestUuid()