Fixes Correlation ID support, adds ability to set arbitrary Correlation IDs

Author: dustindoiron

src/VladimirYuldashev/LaravelQueueRabbitMQ/Queue/Jobs/RabbitMQJob.php

@@ -132,11 +132,26 @@ private function setAttempts($count)
     /**
      * Get the job identifier.
      *
-     * @return string
+     * @return string|false
      */
     public function getJobId()
     {
-        return $this->message->get('correlation_id');
+        if ($this->message->has('correlation_id') === true) {
+            return $this->message->get('correlation_id');
+        }
+
+        return false;
     }
 
+    /**
+     * Sets the job identifier.
+     *
+     * @param string $id
+     *
+     * @return void
+     */
+    public function setJobId($id)
+    {
+        $this->connection->setCorrelationid($id);
+    }
 }

src/VladimirYuldashev/LaravelQueueRabbitMQ/Queue/RabbitMQQueue.php

@@ -33,6 +33,11 @@ class RabbitMQQueue extends Queue implements QueueContract
      */
     private $attempts;
 
+    /**
+     * @var string
+     */
+    private $correlationId;
+
     /**
      * @param AMQPStreamConnection $amqpConnection
      * @param array                $config
@@ -93,7 +98,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
 
         // push job to a queue
         $message = new AMQPMessage($payload, $headers);
-        $this->message->set('correlation_id', uniqid());
+        $message->set('correlation_id', $this->correlationId ?: uniqid());
 
         // push task to a queue
         $this->channel->basic_publish($message, $exchange, $queue);
@@ -249,4 +254,16 @@ public function setAttempts($count)
     {
         $this->attempts = $count;
     }
-}
+
+    /**
+     * Sets the correlation id for a message to be published
+     *
+     * @param string $id
+     *
+     * @return void
+     */
+    public function setCorrelationId($id)
+    {
+        $this->correlationId = $id;
+    }
+}