added: possibility to add custom jobClass to the config.

Author: adm-bome

src/Consumer.php

@@ -74,17 +74,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/Queue/RabbitMQQueue.php

@@ -228,9 +228,11 @@ public function pop($queue = null)
         try {
             $queue = $this->getQueue($queue);
 
+            $job = $this->getJobClass();
+
             /** @var AMQPMessage|null $message */
             if ($message = $this->channel->basic_get($queue)) {
-                return $this->currentJob = new RabbitMQJob(
+                return $this->currentJob = new $job(
                     $this->container,
                     $this,
                     $message,
@@ -272,6 +274,21 @@ public function getChannel(): AMQPChannel
         return $this->channel;
     }
 
+    /**
+     * 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
+     *
+     * @return array|\ArrayAccess|mixed
+     * @throws \Throwable
+     */
+    public function getJobClass()
+    {
+        $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));
+
+        return $job;
+    }
+
     /**
      * Gets a queue/destination, by default the queue option set on the connection.
      *