Merge branch 'feature/use-custom-job-class'

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,
+           ],
        ],
 
        /*
@@ -162,6 +165,92 @@ 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 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 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' => [
+    // ...
+
+    'rabbitmq' => [
+        // ...
+
+        'options' => [
+            'queue' => [
+                // ...
+
+                'job' => \App\Queue\Jobs\RabbitMQJob::class,
+            ],
+        ],
+    ],
+
+    // ...    
+],
+```
+
+An example of your own job class:
+
+```php
+<?php
+
+namespace App\Queue\Jobs;
+
+use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob as BaseJob;
+
+class RabbitMQJob extends BaseJob
+{
+
+    /**
+     * Fire the job.
+     *
+     * @return void
+     */
+    public function fire()
+    {
+        $payload = $this->payload();
+
+        $class = WhatheverClassNameToExecute::class;
+        $method = 'handle';
+
+        ($this->instance = $this->resolve($class))->{$method}($this, $payload);
+    }
+}
+
+```
+
+Or maybe you want to add extra properties to the payload:
+
+```php
+<?php
+
+namespace App\Queue\Jobs;
+
+use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob as BaseJob;
+
+class RabbitMQJob extends BaseJob
+{
+   /**
+     * Get the decoded body of the job.
+     *
+     * @return array
+     */
+    public function payload()
+    {
+        return [
+            'job'  => 'WhatheverFullyQualifiedClassNameToExecute@handle',
+            'data' => json_decode($this->getRawBody(), true)
+        ];
+    }
+}
+```
+
 ## 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/Consumer.php

@@ -12,7 +12,6 @@
 use PhpAmqpLib\Message\AMQPMessage;
 use Symfony\Component\Debug\Exception\FatalThrowableError;
 use Throwable;
-use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue;
 
 class Consumer extends Worker
@@ -74,17 +73,19 @@ public function daemon($connectionName, $queue, WorkerOptions $options): void
             null
         );
 
+        $jobClass = $connection->getJobClass();
+
         $this->channel->basic_consume(
             $queue,
             $this->consumerTag,
             false,
             false,
             false,
             false,
-            function (AMQPMessage $message) use ($connection, $options, $connectionName, $queue): void {
+            function (AMQPMessage $message) use ($connection, $options, $connectionName, $queue, $jobClass): void {
                 $this->gotJob = true;
 
-                $job = new RabbitMQJob(
+                $job = new $jobClass(
                     $this->container,
                     $connection,
                     $message,

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()));
             }
         });