Merge pull request #1 from pdffiller/feature/topics

Added topics supported

Author: osadchyi-s

README.md

@@ -53,6 +53,51 @@ And redis connection in `config/database.php`
 
 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
 
+Also you can use additional features.
+
+### Topics
+Topic help you to put a job to different queues. 
+First, you must to create a subscription. You can use pattern for name of topics. 
+Symbol `*` - one word, `#` - few words divided by point `.`. 
+Examples: `first.second.*`, `*.second.*`, `#.third`.
+
+```php
+/**
+ * Subscribe
+ */
+
+\Queue::subscribe('*.*.test', 'queue1');
+
+\Queue::subscribe('this.*.test', 'queue2');
+
+\Queue::subscribe('this.*.orange', 'queue3');
+
+```
+
+Than you can put job to all subscribers.
+
+```php
+/**
+ * Put job to few queues
+ */
+\Queue::pushToSubscriber('this.is.test', TestQless::class, ['test' => 'test']);
+// Push job to queue1 and queue2, but not to queue3
+
+```
+
+### 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
+worker runs it. Recurring jobs are specified much like other jobs:
+
+```php
+/**
+ * Recurring Job
+ */
+ 
+\Queue::recur($intervalInSeconds, $jobClass, $data, $queueName); 
+
+```
 
 ## Testing
 

composer.json

@@ -15,7 +15,8 @@
         }
     ],
     "require": {
-        "pdffiller/qless-php": "^3.1"
+        "ext-json": "*",
+        "pdffiller/qless-php": "3.3.0"
     },
     "require-dev": {
         "illuminate/events": "5.6.*|5.7.*",

src/LaravelQlessServiceProvider.php

@@ -3,8 +3,12 @@
 namespace LaravelQless;
 
 use Illuminate\Queue\QueueManager;
+use Illuminate\Support\Facades\Config;
 use Illuminate\Support\ServiceProvider;
 use LaravelQless\Queue\QlessConnector;
+use LaravelQless\Topics\QlessTopic;
+use Qless\Client;
+use Qless\Topics\Topic;
 
 /**
  * Class LaravelQlessServiceProvider
@@ -23,7 +27,7 @@ public function boot()
         $queue = $this->app['queue'];
 
         $queue->addConnector('qless', function () {
-            return new QlessConnector($this->app['events']);
+            return new QlessConnector;
         });
     }
 }

src/Queue/QlessQueue.php

@@ -7,6 +7,7 @@
 use Illuminate\Queue\Queue;
 use LaravelQless\Job\QlessJob;
 use Qless\Client;
+use Qless\Topics\Topic;
 
 /**
  * Class QlessQueue
@@ -47,22 +48,6 @@ public function __construct(Client $connect, array $config)
         $this->config = $config;
     }
 
-    /**
-     * @return Client
-     */
-    private function getConnection(): Client
-    {
-        return $this->connect;
-    }
-
-    /**
-     * @return array
-     */
-    private function getConfig(): array
-    {
-        return $this->config;
-    }
-
     /**
      * Get the size of the queue.
      *
@@ -123,7 +108,11 @@ public function push($job, $data = '', $queueName = null)
      */
     public function later($delay, $job, $data = '', $queueName = null)
     {
-        return $this->pushRaw($this->makePayload($job, $data, ['timeout' => $delay]), $queueName, ['timeout' => $delay]);
+        return $this->pushRaw(
+            $this->makePayload($job, $data, ['timeout' => $delay]),
+            $queueName,
+            ['timeout' => $delay]
+        );
     }
 
     /**
@@ -171,6 +160,56 @@ public function pop($queueName = null)
         );
     }
 
+    /**
+     * @param string $topic
+     * @param string|null $queueName
+     * @return bool
+     */
+    public function subscribe(string $topic, string $queueName = null): bool
+    {
+        $queueName = $queueName ?? $this->defaultQueue;
+
+        /** @var \Qless\Queues\Queue $queue */
+        $queue = $this->getConnection()->queues[$queueName];
+
+        return $queue->subscribe($topic);
+    }
+
+    /**
+     * @param string $topic
+     * @param string|null $queueName
+     * @return bool
+     */
+    public function unSubscribe(string $topic, string $queueName = null): bool
+    {
+        $queueName = $queueName ?? $this->defaultQueue;
+
+        /** @var \Qless\Queues\Queue $queue */
+        $queue = $this->getConnection()->queues[$queueName];
+
+        return $queue->unSubscribe($topic);
+    }
+
+    /**
+     * @param string $topic
+     * @param string $job
+     * @param array $data
+     * @param array $options
+     * @return array
+     */
+    public function pushToTopic(string $topic, string $job, array $data = [], array $options = []): array
+    {
+        $topic = new Topic($topic, $this->getConnection());
+
+        return $topic->put(
+            $job,
+            $data,
+            null,
+            $options['timeout'] ?? null,
+            $options['maxTries'] ?? null
+        );
+    }
+
     /**
      * @param string $job
      * @param mixed|string $data
@@ -218,4 +257,12 @@ public function setConnectionName($name): self
 
         return $this;
     }
+
+    /**
+     * @return Client
+     */
+    private function getConnection(): Client
+    {
+        return $this->connect;
+    }
 }

tests/Queue/QlessQueueTest.php

@@ -27,22 +27,76 @@ public function testPushPop()
     {
         $queue = $this->getQueue();
 
-        $jobId = $queue->push(Job::class, ['firstKey' => 'firstValue'], 'test_job');
+        $jobId = $queue->push(Job::class, ['firstKey' => 'firstValue'], 'test_push_pop');
 
-        $job = $queue->pop('test_job');
+        $job = $queue->pop('test_push_pop');
 
         $job->fire();
 
         $this->assertInstanceOf(QlessJob::class, $job);
 
         $this->assertEquals($jobId, $job->getJobId());
 
-        $this->assertEquals($job->getQueue(), 'test_job');
+        $this->assertEquals($job->getQueue(), 'test_push_pop');
 
         $this->assertEquals($job->getName(), Job::class);
 
         $this->assertEquals($job->getData(), ['firstKey' => 'firstValue']);
     }
+
+    public function testSubscribe()
+    {
+        $queue = $this->getQueue();
+
+        for ($i = 1; $i<=3; $i++) {
+            $queue->push(Job::class, ['firstKey' => 'firstValue'], 'qs_' . $i);
+            $job = $queue->pop('qs_' . $i);
+            $job->fire();
+        }
+
+        $queue->subscribe('developing#', 'qs_1');
+        $queue->subscribe('*.test.*', 'qs_2');
+        $queue->subscribe('*.*.awesome', 'qs_3');
+
+        $queue->pushToTopic('developing.is.awesome', Job::class, ['key' => 'value']);
+
+        $job1 = $queue->pop('qs_1');
+        $job1->fire();
+        $job2 = $queue->pop('qs_2');
+        $job3 = $queue->pop('qs_3');
+        $job3->fire();
+
+        $this->assertEquals($job1->getName(), Job::class);
+        $this->assertEquals($job2, null);
+        $this->assertEquals($job3->getName(), Job::class);
+
+        $this->assertEquals($job1->getData(), ['key' => 'value']);
+        $this->assertEquals($job3->getData(), ['key' => 'value']);
+    }
+    public function testUnSubscribe()
+    {
+        $queue = $this->getQueue();
+
+        for ($i = 1; $i<=2; $i++) {
+            $queue->push(Job::class, ['firstKey' => 'firstValue'], 'qu_' . $i);
+            $job = $queue->pop('qu_' . $i);
+            $job->fire();
+        }
+
+        $queue->subscribe('developing.*.*', 'qu_1');
+        $queue->subscribe('*.*.awesome', 'qu_2');
+
+        $queue->unSubscribe('*.*.awesome', 'qu_2');
+
+        $queue->pushToTopic('developing.is.awesome', Job::class, ['key' => 'value']);
+
+        $job1 = $queue->pop('qu_1');
+        $job1->fire();
+        $job2 = $queue->pop('qu_2');
+
+        $this->assertEquals($job1->getName(), Job::class);
+        $this->assertEquals($job2, null);
+    }
 
     public function testSize()
     {