wip

Author: vyuldashev

README.md

@@ -61,6 +61,9 @@ Add connection to `config/queue.php`:
                'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
                'passphrase' => env('RABBITMQ_SSL_PASSPHRASE', null),
            ],
+           'queue' => [
+               'job' => VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob::class,
+           ],
        ],
 
        /*
@@ -161,15 +164,17 @@ When you want to instruct RabbitMQ to reroute failed messages to a exchange or a
     // ...    
 ],
 ```
+
 ### Use your own RabbitMQJob class
 Sometimes you have to work with messages published by another application.  
-Those messages probably won't respect the laravel QueueApi payload schema.
-The problem with these messages is that, laravel workers won't be able to determine the actual job or class to execute. 
+Those messages probably won't respect Laravel's job payload schema.
+The problem with these messages is that, Laravel workers won't be able to determine the actual job or class to execute. 
 
-You can extend the build-in `RabbitMQJob::class` and within the queue connection config, you can add your own class.
+You can extend the build-in `RabbitMQJob::class` and within the queue connection config, you can define your own class.
 When you specify an `job` key in the config, with your own class name, every message retrieved from the broker will get wrapped by your own class.  
 
 An example for the config:
+
 ```php
 'connections' => [
     // ...
@@ -189,7 +194,9 @@ An example for the config:
     // ...    
 ],
 ```
+
 An example of your own job class:
+
 ```php
 <?php
 
@@ -217,7 +224,9 @@ class RabbitMQJob extends BaseJob
 }
 
 ```
-Or maybe you want to add extra properties to the payload.
+
+Or maybe you want to add extra properties to the payload:
+
 ```php
 <?php
 
@@ -241,6 +250,7 @@ class RabbitMQJob extends BaseJob
     }
 }
 ```
+
 ## Laravel Usage
 
 Once you completed the configuration you can use Laravel Queue API. If you used other queue drivers you do not need to change anything else. If you do not know how to use Queue API, please refer to the official Laravel documentation: http://laravel.com/docs/queues

config/rabbitmq.php

@@ -29,6 +29,9 @@
             'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
             'passphrase' => env('RABBITMQ_SSL_PASSPHRASE', null),
         ],
+        'queue' => [
+            'job' => VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob::class,
+        ],
     ],
 
     /*

src/Horizon/RabbitMQQueue.php

@@ -74,7 +74,7 @@ public function later($delay, $job, $data = '', $queue = null)
     public function pop($queue = null)
     {
         return tap(parent::pop($queue), function ($result) use ($queue): void {
-            if ($result instanceof RabbitMQJob) {
+            if (is_a($result, RabbitMQJob::class, true)) {
                 $this->event($this->getQueue($queue), new JobReserved($result->getRawBody()));
             }
         });

src/Queue/RabbitMQQueue.php

@@ -10,12 +10,15 @@
 use Illuminate\Queue\Queue;
 use Illuminate\Support\Arr;
 use Illuminate\Support\Str;
+use JsonException;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Exception\AMQPProtocolChannelException;
 use PhpAmqpLib\Exchange\AMQPExchangeType;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
+use Throwable;
+use Throwabler;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob;
 
 class RabbitMQQueue extends Queue implements QueueContract
@@ -101,7 +104,7 @@ public function size($queue = null): int
     {
         $queue = $this->getQueue($queue);
 
-        if (! $this->isQueueExists($queue)) {
+        if (!$this->isQueueExists($queue)) {
             return 0;
         }
 
@@ -172,7 +175,7 @@ public function laterRaw($delay, $payload, $queue = null, $attempts = 0)
             return $this->pushRaw($payload, $queue, ['delay' => $delay, 'attempts' => $attempts]);
         }
 
-        $destination = $this->getQueue($queue).'.delay.'.$ttl;
+        $destination = $this->getQueue($queue) . '.delay.' . $ttl;
 
         $this->declareQueue($destination, true, false, $this->getDelayQueueArguments($this->getQueue($queue), $ttl));
 
@@ -191,7 +194,7 @@ public function laterRaw($delay, $payload, $queue = null, $attempts = 0)
      */
     public function bulk($jobs, $data = '', $queue = null): void
     {
-        foreach ((array) $jobs as $job) {
+        foreach ((array)$jobs as $job) {
             $this->bulkRaw($this->createPayload($job, $queue, $data), $queue, ['job' => $job]);
         }
 
@@ -221,7 +224,7 @@ public function bulkRaw(string $payload, $queue = null, array $options = [])
     /**
      * {@inheritdoc}
      *
-     * @throws Exception
+     * @throws Throwable
      */
     public function pop($queue = null)
     {
@@ -244,7 +247,6 @@ public function pop($queue = null)
             // If there is not exchange or queue AMQP will throw exception with code 404
             // We need to catch it and return null
             if ($exception->amqp_reply_code === 404) {
-
                 // Because of the channel exception the channel was closed and removed.
                 // We have to open a new channel. Because else the worker(s) are stuck in a loop, without processing.
                 $this->channel = $this->connection->channel();
@@ -275,16 +277,20 @@ public function getChannel(): AMQPChannel
     }
 
     /**
-     * Gets the Job class from config or returns the default job class
-     * when the job class does not extend the default job class an exception is thrown.
+     * Job class to use.
      *
      * @return string
      * @throws \Throwable
      */
-    public function getJobClass()
+    public function getJobClass(): string
     {
         $job = Arr::get($this->options, 'job', RabbitMQJob::class);
-        throw_if(! is_a($job, RabbitMQJob::class, true), Exception::class, sprintf('Class %s must extend: %s', $job, RabbitMQJob::class));
+
+        throw_if(
+            !is_a($job, RabbitMQJob::class, true),
+            Exception::class,
+            sprintf('Class %s must extend: %s', $job, RabbitMQJob::class)
+        );
 
         return $job;
     }
@@ -295,7 +301,7 @@ public function getJobClass()
      * @param null $queue
      * @return string
      */
-    public function getQueue($queue = null)
+    public function getQueue($queue = null): string
     {
         return $queue ?: $this->default;
     }
@@ -336,8 +342,13 @@ public function isExchangeExists(string $exchange): bool
      * @param array $arguments
      * @return void
      */
-    public function declareExchange(string $name, string $type = AMQPExchangeType::DIRECT, bool $durable = true, bool $autoDelete = false, array $arguments = []): void
-    {
+    public function declareExchange(
+        string $name,
+        string $type = AMQPExchangeType::DIRECT,
+        bool $durable = true,
+        bool $autoDelete = false,
+        array $arguments = []
+    ): void {
         if ($this->isExchangeDeclared($name)) {
             return;
         }
@@ -364,7 +375,7 @@ public function declareExchange(string $name, string $type = AMQPExchangeType::D
      */
     public function deleteExchange(string $name, bool $unused = false): void
     {
-        if (! $this->isExchangeExists($name)) {
+        if (!$this->isExchangeExists($name)) {
             return;
         }
 
@@ -378,7 +389,7 @@ public function deleteExchange(string $name, bool $unused = false): void
      * Checks if the given queue already present/defined in RabbitMQ.
      * Returns false when when the queue is missing.
      *
-     * @param string $name
+     * @param string|null $name
      * @return bool
      * @throws AMQPProtocolChannelException
      */
@@ -409,8 +420,12 @@ public function isQueueExists(string $name = null): bool
      * @param array $arguments
      * @return void
      */
-    public function declareQueue(string $name, bool $durable = true, bool $autoDelete = false, array $arguments = []): void
-    {
+    public function declareQueue(
+        string $name,
+        bool $durable = true,
+        bool $autoDelete = false,
+        array $arguments = []
+    ): void {
         if ($this->isQueueDeclared($name)) {
             return;
         }
@@ -437,7 +452,7 @@ public function declareQueue(string $name, bool $durable = true, bool $autoDelet
      */
     public function deleteQueue(string $name, bool $if_unused = false, bool $if_empty = false): void
     {
-        if (! $this->isQueueExists($name)) {
+        if (!$this->isQueueExists($name)) {
             return;
         }
 
@@ -468,7 +483,7 @@ public function bindQueue(string $queue, string $exchange, string $routingKey =
     /**
      * Purge the queue of messages.
      *
-     * @param string $queue
+     * @param string|null $queue
      * @return void
      */
     public function purge(string $queue = null): void
@@ -509,6 +524,7 @@ public function reject(RabbitMQJob $job, bool $requeue = false): void
      * @param $payload
      * @param int $attempts
      * @return array
+     * @throws JsonException
      */
     protected function createMessage($payload, int $attempts = 0): array
     {
@@ -517,7 +533,7 @@ protected function createMessage($payload, int $attempts = 0): array
             'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
         ];
 
-        if ($correlationId = json_decode($payload, true)['id'] ?? null) {
+        if ($correlationId = json_decode($payload, true, 512)['id'] ?? null) {
             $properties['correlation_id'] = $correlationId;
         }
 
@@ -542,12 +558,12 @@ protected function createMessage($payload, int $attempts = 0): array
     /**
      * Create a payload array from the given job and data.
      *
-     * @param object|string $job
+     * @param string|object $job
      * @param string $queue
-     * @param string $data
+     * @param mixed $data
      * @return array
      */
-    protected function createPayloadArray($job, $queue, $data = '')
+    protected function createPayloadArray($job, $queue, $data = ''): array
     {
         return array_merge(parent::createPayloadArray($job, $queue, $data), [
             'id' => $this->getRandomId(),
@@ -572,7 +588,7 @@ protected function getRandomId(): string
      */
     public function close(): void
     {
-        if ($this->currentJob && ! $this->currentJob->isDeletedOrReleased()) {
+        if ($this->currentJob && !$this->currentJob->isDeletedOrReleased()) {
             $this->reject($this->currentJob, true);
         }
 
@@ -632,7 +648,7 @@ protected function getDelayQueueArguments(string $destination, int $ttl): array
      */
     protected function isPrioritizeDelayed(): bool
     {
-        return boolval(Arr::get($this->options, 'prioritize_delayed') ?: false);
+        return (bool)(Arr::get($this->options, 'prioritize_delayed') ?: false);
     }
 
     /**
@@ -645,13 +661,13 @@ protected function isPrioritizeDelayed(): bool
      */
     protected function getQueueMaxPriority(): int
     {
-        return intval(Arr::get($this->options, 'queue_max_priority') ?: 2);
+        return (int)(Arr::get($this->options, 'queue_max_priority') ?: 2);
     }
 
     /**
      * Get the exchange name, or &null; as default value.
      *
-     * @param string $exchange
+     * @param string|null $exchange
      * @return string|null
      */
     protected function getExchange(string $exchange = null): ?string
@@ -679,7 +695,8 @@ protected function getRoutingKey(string $destination): string
      */
     protected function getExchangeType(?string $type = null): string
     {
-        return @constant(AMQPExchangeType::class.'::'.Str::upper($type ?: Arr::get($this->options, 'exchange_type') ?: 'direct')) ?: AMQPExchangeType::DIRECT;
+        return @constant(AMQPExchangeType::class . '::' . Str::upper($type ?: Arr::get($this->options,
+                'exchange_type') ?: 'direct')) ?: AMQPExchangeType::DIRECT;
     }
 
     /**
@@ -689,7 +706,7 @@ protected function getExchangeType(?string $type = null): string
      */
     protected function isRerouteFailed(): bool
     {
-        return boolval(Arr::get($this->options, 'reroute_failed') ?: false);
+        return (bool)(Arr::get($this->options, 'reroute_failed') ?: false);
     }
 
     /**
@@ -746,10 +763,13 @@ protected function isQueueDeclared(string $name): bool
      * @return void
      * @throws AMQPProtocolChannelException
      */
-    protected function declareDestination(string $destination, ?string $exchange = null, string $exchangeType = AMQPExchangeType::DIRECT): void
-    {
+    protected function declareDestination(
+        string $destination,
+        ?string $exchange = null,
+        string $exchangeType = AMQPExchangeType::DIRECT
+    ): void {
         // When a exchange is provided and no exchange is present in RabbitMQ, create an exchange.
-        if ($exchange && ! $this->isExchangeExists($exchange)) {
+        if ($exchange && !$this->isExchangeExists($exchange)) {
             $this->declareExchange($exchange, $exchangeType);
         }