vyuldashev#225 laravel horizon support

Author: vyuldashev

composer.json

@@ -1,6 +1,6 @@
 {
   "name": "vladimir-yuldashev/laravel-queue-rabbitmq",
-  "description": "RabbitMQ driver for Laravel Queue",
+  "description": "RabbitMQ driver for Laravel Queue. Supports Laravel Horizon",
   "license": "MIT",
   "authors": [
     {

src/Horizon/RabbitMQQueue.php

@@ -23,7 +23,7 @@ class RabbitMQQueue extends BaseRabbitMQQueue
     /**
      * Get the number of queue jobs that are ready to process.
      *
-     * @param  string|null  $queue
+     * @param  string|null $queue
      * @return int
      */
     public function readyNow($queue = null): int
@@ -45,7 +45,7 @@ public function pushRaw($payload, $queueName = null, array $options = [])
         $payload = (new JobPayload($payload))->prepare($this->lastPushed)->value;
 
         return tap(parent::pushRaw($payload, $queueName, $options), function () use ($queueName, $payload) {
-            $this->event($queueName ?: $this->queueName, new JobPushed($payload));
+            $this->event($this->getQueueName($queueName), new JobPushed($payload));
         });
     }
 
@@ -55,7 +55,7 @@ public function later($delay, $job, $data = '', $queueName = null)
         $payload = (new JobPayload($this->createPayload($job, $data)))->prepare($job)->value;
 
         return tap(parent::pushRaw($payload, $queueName, ['delay' => $this->secondsUntil($delay)]), function () use ($payload, $queueName) {
-            $this->event($queueName ?: $this->queueName, new JobPushed($payload));
+            $this->event($this->getQueueName($queueName), new JobPushed($payload));
         });
     }
 
@@ -87,7 +87,7 @@ public function release($delay, $job, $data, $queue, $attempts = 0)
      */
     public function deleteReserved($queueName, $job): void
     {
-        $this->event($queueName ?: $this->queueName, new JobDeleted($job, $job->getRawBody()));
+        $this->event($this->getQueueName($queueName), new JobDeleted($job, $job->getRawBody()));
     }
 
     /**

src/Queue/RabbitMQQueue.php

@@ -185,7 +185,7 @@ public function getContext(): AmqpContext
      */
     protected function declareEverything(string $queueName = null): array
     {
-        $queueName = $queueName ?: $this->queueName;
+        $queueName = $this->getQueueName($queueName);
         $exchangeName = $this->exchangeOptions['name'] ?: $queueName;
 
         $topic = $this->context->createTopic($exchangeName);
@@ -235,6 +235,27 @@ protected function declareEverything(string $queueName = null): array
         return [$queue, $topic];
     }
 
+    protected function getQueueName($queueName = null) {
+        return $queueName ?: $this->queueName;
+    }
+
+    protected function createPayloadArray($job, $queue, $data = '')
+    {
+        return array_merge(parent::createPayloadArray($job, $queue, $data), [
+            'id' => $this->getRandomId(),
+        ]);
+    }
+
+    /**
+     * Get a random ID string.
+     *
+     * @return string
+     */
+    protected function getRandomId(): string
+    {
+        return Str::random(32);
+    }
+
     /**
      * @param string $action
      * @param \Throwable $e
@@ -255,21 +276,4 @@ protected function reportConnectionError($action, \Throwable $e)
         // Sleep so that we don't flood the log file
         sleep($this->sleepOnError);
     }
-
-    protected function createPayloadArray($job, $queue, $data = '')
-    {
-        return array_merge(parent::createPayloadArray($job, $queue, $data), [
-            'id' => $this->getRandomId(),
-        ]);
-    }
-
-    /**
-     * Get a random ID string.
-     *
-     * @return string
-     */
-    protected function getRandomId(): string
-    {
-        return Str::random(32);
-    }
 }