Support dispatch

Author: osadchyi-s

README.md

@@ -31,7 +31,7 @@ Also you can set Qless queue as default in  `config/queue.php`
     'default' => env('QUEUE_DRIVER', 'qless'),
 ```
 
-And redis connection in `config/database.php`
+Redis connection in `config/database.php`
 
 ```php
     'redis' => [
@@ -49,6 +49,23 @@ And redis connection in `config/database.php`
     ],
 ```
 
+And add Laravel Qless service provider to app.php
+
+```php
+<?php
+
+return [
+    //...
+    'providers' => [
+        //...
+        /**
+         * Qless
+         */
+         LaravelQless\LaravelQlessServiceProvider::class,
+    ]
+];
+```
+
 ## 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
@@ -80,7 +97,7 @@ Than you can put job to all subscribers.
 /**
  * Put job to few queues
  */
-\Queue::pushToSubscriber('this.is.test', TestQless::class, ['test' => 'test']);
+\Queue::pushToTopic('this.is.test', TestQless::class, ['test' => 'test']);
 // Push job to queue1 and queue2, but not to queue3
 
 ```

config/queue.php

@@ -1,10 +1,12 @@
 <?php
 
 return [
+    'default' => 'qless',
+
     'connections' => [
         'qless' => [
             'driver' => 'qless',
-            'connection' => 'default',
+            'connection' => 'qless',
             'queue' => 'default',
             'redis_connection' => 'qless',
         ],

examples/Job/ExampleJob.php

@@ -2,12 +2,11 @@
 
 namespace LaravelQless\Examples\Job;
 
-use LaravelQless\Contracts\QlessJob;
+use LaravelQless\Job\AbstractJob;
 use Qless\Jobs\BaseJob;
 
-class ExampleJob implements QlessJob
+class ExampleJob extends AbstractJob
 {
-
     /**
      * Execute the job.
      *

src/Job/AbstractJob.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace LaravelQless\Job;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Contracts\Support\Arrayable;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Qless\Jobs\BaseJob;
+use LaravelQless\Contracts\QlessJob;
+
+/**
+ * Class AbstractJob
+ * @package LaravelQless\Job
+ */
+abstract class AbstractJob implements QlessJob, ShouldQueue, Arrayable
+{
+    use Dispatchable, Queueable;
+    /**
+     * @var array
+     */
+    protected $data = [];
+
+    public function __construct(array $data = [])
+    {
+        $this->data = $data;
+    }
+
+    /**
+     * @param BaseJob $job
+     * @return mixed
+     */
+    abstract public function perform(BaseJob $job);
+
+    /**
+     * @return array
+     */
+    public function toArray(): array
+    {
+        return (array) $this->data;
+    }
+}

src/Queue/QlessConnector.php

@@ -21,7 +21,7 @@ class QlessConnector implements ConnectorInterface
     */
     public function connect(array $config): QlessQueue
     {
-         $redisConnection = array_get($config, 'redis_connection', 'default');
+         $redisConnection = array_get($config, 'redis_connection', 'qless');
 
          $redisConfig = Config::get('database.redis.' . $redisConnection, []);
 

src/Queue/QlessQueue.php

@@ -101,7 +101,7 @@ public function pushRaw($payload, $queueName = null, array $options = [])
      */
     public function push($job, $data = '', $queueName = null)
     {
-        return $this->pushRaw($this->makePayload($job, $data), $queueName);
+        return $this->pushRaw($this->makePayload($job, (array) $data), $queueName);
     }
 
     /**
@@ -240,19 +240,26 @@ public function pushToTopic(string $topicName, string $job, array $data = [], ar
     }
 
     /**
-     * @param string $job
+     * @param string|object $job
      * @param mixed|string $data
      * @param array $options
      * @return string
      */
-    protected function makePayload(string $job, $data, $options = []): string
+    protected function makePayload($job, $data = [], $options = []): string
     {
+        if (is_object($job)) {
+            $displayName = get_class($job);
+            $data = array_merge($job->toArray(), $data);
+        } else {
+            $displayName = explode('@', $job)[0];
+        }
+
         $qlessOptions = $data[self::JOB_OPTIONS_KEY] ?? [];
         $data[self::JOB_OPTIONS_KEY] = array_merge($qlessOptions, $options);
 
         $payload = json_encode([
-            'displayName' => explode('@', $job)[0],
-            'job' => $job,
+            'displayName' => $displayName,
+            'job' => is_string($job) ? $job : $displayName,
             'data' => $data,
         ]);
 

tests/Queue/QlessConnectorTest.php renamed to tests/Queue/ConnectorTest.php

@@ -8,7 +8,7 @@
 use LaravelQless\Queue\QlessConnector;
 use LaravelQless\Queue\QlessQueue;
 
-class QlessConnectorTest extends TestCase
+class ConnectorTest extends TestCase
 {
     public function testShouldImplementConnectorInterface()
     {
@@ -20,7 +20,8 @@ public function testConnect()
     {
         $connector = new QlessConnector();
         $queue = $connector->connect([
-            'redis_connection' => 'qless'
+            'redis_connection' => 'qless',
+            'connection' => 'qless',
         ]);
         $this->assertInstanceOf(QlessQueue::class, $queue);
     }

tests/Queue/HandlerTest.php

@@ -30,7 +30,8 @@ protected function getQueue()
                 'port' => REDIS_PORT,
             ]),
             [
-                'queue' => 'test_qless_queue'
+                'queue' => 'test_qless_queue',
+                'connection' => 'qless',
             ]
         );
 

tests/Queue/Job.php

@@ -2,10 +2,10 @@
 
 namespace LaravelQless\Tests\Queue;
 
-use LaravelQless\Contracts\QlessJob;
+use LaravelQless\Job\AbstractJob;
 use Qless\Jobs\BaseJob;
 
-class Job implements QlessJob
+class Job extends AbstractJob
 {
     public function perform(BaseJob $job)
     {

tests/Queue/QlessQueueTest.php renamed to tests/Queue/QueueTest.php

@@ -2,16 +2,18 @@
 
 namespace LaravelQless\Tests\Queue;
 
+use Illuminate\Queue\QueueManager;
 use LaravelQless\Contracts\JobHandler;
 use LaravelQless\Handler\DefaultHandler;
 use LaravelQless\Job\QlessJob;
+use LaravelQless\Queue\QlessConnector;
 use LaravelQless\Queue\QlessQueue;
 use Orchestra\Testbench\TestCase;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Queue;
 use Qless\Client;
 
-class QlessQueueTest extends TestCase
+class QueueTest extends TestCase
 {
     /**
      * @throws \ReflectionException
@@ -173,6 +175,30 @@ public function testJobOptions()
         $this->assertEquals(['tag1', 'tag_second'], $job->getData()['tags']);
     }
 
+    public function testDispatchJob()
+    {
+        $this->setEnv();
+
+        $queueName = str_random();
+
+        $job = new Job(['dispatch' => 'work']);
+
+        $dispatch = dispatch($job)->onQueue($queueName)->onConnection('qless');
+        unset($dispatch);
+
+        $queue = $this->getQueue();
+
+        $job = $queue->pop($queueName);
+
+        $this->assertNotEmpty($job);
+
+        $job->fire();
+
+        $data = $job->getData();
+
+        $this->assertEquals('work', $data['dispatch']);
+    }
+
     protected function getQueue()
     {
         $queue = new QlessQueue(
@@ -181,7 +207,8 @@ protected function getQueue()
                 'port' => REDIS_PORT,
             ]),
             [
-                'queue' => 'test_qless_queue'
+                'queue' => 'test_qless_queue',
+                'connection' => 'qless',
             ]
         );
 
@@ -190,6 +217,36 @@ protected function getQueue()
         return $queue;
     }
 
+
+    protected function setEnv()
+    {
+        $this->app['config']->set('queue.default', 'qless');
+        $qlessConfig = [
+            'driver' => 'qless',
+            'connection' => 'qless',
+            'queue' => 'default',
+            'redis_connection' => 'qless',
+        ];
+        $this->app['config']->set('queue.connections.qless', $qlessConfig);
+
+        $redisConfig = [
+            'host' => REDIS_HOST,
+            'password' => null,
+            'port' => REDIS_PORT,
+            'database' => 0,
+        ];
+
+        $this->app['config']->set('database.redis.qless', $redisConfig);
+
+        $queueManager = new QueueManager($this->app);
+        $queueManager->addConnector('qless', function () {
+            return new QlessConnector;
+        });
+        $queueManager->setDefaultDriver('qless');
+
+        $this->app['queue'] = $queueManager;
+    }
+
     protected function getApplicationProviders($app)
     {
         $app->bind(JobHandler::class, DefaultHandler::class);