Custom handler

Author: osadchyi-s

README.md

@@ -18,7 +18,7 @@ Setup connection in `config/queue.php`
         // ...
         'qless' => [
             'driver' => 'qless',
-            'connection' => 'qless',
+            'redis_connection' => 'qless',
             'queue' => 'default',
         ],
         // ...    
@@ -85,6 +85,56 @@ Than you can put job to all subscribers.
 
 ```
 
+### Custom Handler
+Job Handler helps you to create a custom route for jobs.
+
+Custom Handler example:
+
+```php
+
+    class CustomHandler implements JobHandler
+    {
+        public function perform(BaseJob $job): void
+        {
+            if ($job->getQueue() === 'queue_name') { // by queue
+                (new QueueNameJob)->perform($job);
+                return;
+            }
+            
+            if ($job->getData()['option_name'] === 'value') { // by payload
+                (new OptionNameJob)->perform($job);
+                return;
+            }
+            
+            if (in_array('tag_name', $job->getTags())) { // by tag
+                (new TagNameJob)->perform($job);
+                return;
+            }
+            
+            // other
+            
+            $job->perform(); // Default
+        }
+    }
+    
+```
+
+You must add to a service provider.
+
+```php
+
+    public function boot()
+    {
+        // other code
+
+        $this->app->bind(JobHandler::class, CustomHandler::class);
+        
+        // other code
+    }
+
+```
+
+
 ### Recurring Jobs
 Sometimes it's not enough simply to schedule one job, but you want to run jobs regularly.
 In particular, maybe you have some batch operation that needs to get run once an hour and you don't care what
@@ -117,6 +167,6 @@ Laravel Qless Queue driver is open-sourced software licensed under the MIT Licen
 See the [`LICENSE.txt`](https://github.com/pdffiller/laravel-queue-qless/blob/master/LICENSE.txt) file for more.
 
 
-© 2018 PDFfiller<br>
+© 2018-2019 PDFfiller<br>
 
 All rights reserved.

src/Contracts/JobHandler.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace LaravelQless\Contracts;
+
+use Qless\Jobs\PerformAwareInterface;
+
+interface JobHandler extends PerformAwareInterface
+{
+
+}

src/Handler/DefaultHandler.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace LaravelQless\Handler;
+
+use LaravelQless\Contracts\JobHandler;
+use Qless\Jobs\BaseJob;
+
+class DefaultHandler implements JobHandler
+{
+    public function perform(BaseJob $job): void
+    {
+        $job->perform();
+    }
+}

src/Job/QlessJob.php

@@ -6,6 +6,7 @@
 use Illuminate\Queue\Jobs\Job;
 use Illuminate\Contracts\Queue\Job as JobContract;
 use Qless\Jobs\BaseJob;
+use LaravelQless\Contracts\JobHandler;
 
 /**
  * Class QlessJob
@@ -38,11 +39,17 @@ class QlessJob extends Job implements JobContract
      */
     protected $released = false;
 
+    /**
+     * @var JobHandler
+     */
+    protected $handler;
+
     public function __construct(BaseJob $job, string $payload, string $connectionName)
     {
         $this->job = $job;
         $this->payload = $payload;
         $this->connectionName = $connectionName;
+        $this->handler = app()->make(JobHandler::class);
     }
 
     /**
@@ -80,7 +87,7 @@ public function payload()
      */
     public function fire()
     {
-        $this->job->perform();
+        $this->handler->perform($this->job);
     }
 
     /**
@@ -91,9 +98,9 @@ public function fire()
      */
     public function release($delay = 0)
     {
-        //$this->delete();
         $this->released = true;
-        return \Queue::later($delay, $this->job->getKlass(), $this->job->getData(), $this->job->getQueue());
+        return \Queue::connection($this->getConnectionName())
+            ->later($delay, $this->job->getKlass(), $this->job->getData(), $this->job->getQueue());
     }
 
     /**

src/LaravelQlessServiceProvider.php

@@ -3,12 +3,10 @@
 namespace LaravelQless;
 
 use Illuminate\Queue\QueueManager;
-use Illuminate\Support\Facades\Config;
 use Illuminate\Support\ServiceProvider;
+use LaravelQless\Contracts\JobHandler;
+use LaravelQless\Handler\DefaultHandler;
 use LaravelQless\Queue\QlessConnector;
-use LaravelQless\Topics\QlessTopic;
-use Qless\Client;
-use Qless\Topics\Topic;
 
 /**
  * Class LaravelQlessServiceProvider
@@ -29,5 +27,7 @@ public function boot()
         $queue->addConnector('qless', function () {
             return new QlessConnector;
         });
+        
+        $this->app->bindif(JobHandler::class, DefaultHandler::class);
     }
 }

src/Queue/QlessQueue.php

@@ -15,6 +15,8 @@
  */
 class QlessQueue extends Queue implements QueueContract
 {
+    private const WORKER_PREFIX = 'laravel_';
+
     /**
      * @var Client
      */
@@ -136,7 +138,7 @@ public function recur(int $interval, string $job, array $data, ?string $queueNam
      * Pop the next job off of the queue.
      *
      * @param  string  $queueName
-     * @return \Illuminate\Contracts\Queue\Job|null
+     * @return \Illuminate\Contracts\Queue\Job|null|QlessJob
      */
     public function pop($queueName = null)
     {
@@ -145,7 +147,7 @@ public function pop($queueName = null)
         /** @var \Qless\Queues\Queue $queue */
         $queue = $this->getConnection()->queues[$queueName];
 
-        $job = $queue->pop();
+        $job = $queue->pop(self::WORKER_PREFIX . $this->connect->getWorkerName());
 
         if (!$job) {
             return null;

tests/Queue/CustomHandler.php

@@ -0,0 +1,16 @@
+<?php
+
+namespace LaravelQless\Tests\Queue;
+
+use LaravelQless\Contracts\JobHandler;
+use Qless\Jobs\BaseJob;
+
+class CustomHandler implements JobHandler
+{
+    public function perform(BaseJob $job): void
+    {
+        if ($job->getQueue() === 'handler_test') {
+            (new Job)->perform($job);
+        }
+    }
+}

tests/Queue/HandlerTest.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace LaravelQless\Tests\Queue;
+
+use LaravelQless\Contracts\JobHandler;
+use LaravelQless\Queue\QlessQueue;
+use Orchestra\Testbench\TestCase;
+use Qless\Client;
+
+class HandlerTest extends TestCase
+{
+    public function testCustomHandler()
+    {
+        $queue = $this->getQueue();
+
+        $queue->push(\stdClass::class, ['firstKey' => 'firstValue'], 'handler_test');
+
+        $job = $queue->pop('handler_test');
+
+        $job->fire();
+
+        $this->assertEquals($job->getData()['classHandler'], Job::class);
+    }
+
+    protected function getQueue()
+    {
+        return new QlessQueue(
+            new Client([
+                'host' => REDIS_HOST,
+                'port' => REDIS_PORT,
+            ]),
+            [
+                'queue' => 'test_qless_queue'
+            ]
+        );
+    }
+
+    protected function getApplicationProviders($app)
+    {
+        $app->bind(JobHandler::class, CustomHandler::class);
+
+        return $app['config']['app.providers'];
+    }
+}

tests/Queue/Job.php

@@ -9,6 +9,10 @@ class Job implements QlessJob
 {
     public function perform(BaseJob $job)
     {
+        if ($job->getQueue() === 'handler_test') {
+            $job->getData()['classHandler'] = self::class;
+        }
+
         $job->complete();
     }
-}
+}

tests/Queue/QlessConnectorTest.php

@@ -2,6 +2,7 @@
 
 namespace LaravelQless\Tests\Queue;
 
+use LaravelQless\Contracts\JobHandler;
 use Orchestra\Testbench\TestCase;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 use LaravelQless\Queue\QlessConnector;
@@ -36,4 +37,11 @@ protected function getEnvironmentSetUp($app)
             'port' => REDIS_PORT,
         ]);
     }
+
+    protected function getApplicationProviders($app)
+    {
+        $app->bindIf(JobHandler::class, CustomHandler::class);
+
+        return $app['config']['app.providers'];
+    }
 }