Merge pull request #47 from UrosPurtic/master

Add queue source to tasker domain for additional info when loggin tasks

Author: ppavlovic

src/Model/Domain/Task.php

@@ -65,6 +65,10 @@ class Task
      * @var int
      */
     private $startedCount;
+    /**
+     * @var mixed
+     */
+    private $queueSource;
 
     /**
      * Task constructor.
@@ -104,6 +108,7 @@ public function getRawData()
             'ts_started'     => $this->getTsStarted(),
             'exec_time'      => $this->getExecTime(),
             'started_count'  => $this->getStartedCount(),
+            'queue_source'   => $this->getQueueSource(),
         );
     }
 
@@ -367,6 +372,16 @@ public function setStatusDone($execTime=null)
         return $this;
     }
 
+    public function setQueueSource($queueSource)
+    {
+        $this->queueSource = $queueSource;
+    }
+
+    public function getQueueSource()
+    {
+        return $this->queueSource ?: null;
+    }
+
     public function setStatusWaitingForRetry($execTime=null)
     {
         $this->status = \G4\Tasker\Consts::STATUS_WAITING_FOR_RETRY;
@@ -414,7 +429,8 @@ public static function fromData($data)
             ->setStatus((int) $data['status'])
             ->setTsStarted((int) $data['ts_started'])
             ->setExecTime((float) $data['exec_time'])
-            ->setStartedCount((int) $data['started_count']);
+            ->setStartedCount((int) $data['started_count'])
+            ->setQueueSource(isset($data['queue_source']) ? (string) $data['queue_source'] : null );
 
         return $task;
     }

src/Tasker2/Runner.php

@@ -65,12 +65,17 @@ class Runner extends \G4\Tasker\TimerAbstract
 
     const LOG_TYPE = 'rb_worker';
 
-    public function __construct(AMQPMessage $AMQPMessage, TaskRepositoryInterface $taskRepository, array $delayForRetries = [])
-    {
+    public function __construct(
+        AMQPMessage $AMQPMessage,
+        TaskRepositoryInterface $taskRepository,
+        array $delayForRetries = [],
+        string $queue = null
+    ){
         $this->taskRepository = $taskRepository;
         $this->taskData = new \G4\ValueObject\Dictionary(
             json_decode($AMQPMessage->getBody(), true)
         );
+        $this->taskData->add('queue_source', $queue);
         $this->taskDomain = \G4\Tasker\Model\Domain\Task::fromData($this->taskData->getAll());
         $this->taskerExecution = (new \G4\Log\Data\TaskerExecution())->setLogType(self::LOG_TYPE);
         $this->resolver = new RetryAfterResolver($delayForRetries);