Imeplement RemoteQueue class to push jobs

Author: mzur

README.md

@@ -23,6 +23,22 @@ todo
 
 ## Usage
 
+To use the remote queue, configure a queue connection in the `queue.connections` config to use the `remote` driver:
+
+```php
+[
+   'driver' => 'remote',
+   // Default queue of the remote host to push jobs to.
+   'queue' => 'default',
+   // The remote queue API endpoint of the remote host.
+   'url' => 'http://192.168.100.100/api/v1/remote-queue',
+   // Basic auth username to use for authentication on the remote host.
+   'username' => '',
+   // Basic auth password to use for authentication on the remote host.
+   'password' => '',
+]
+```
+
 ```php
 use App\Jobs\MyJob;
 

src/RemoteConnector.php

@@ -2,6 +2,7 @@
 
 namespace Biigle\RemoteQueue;
 
+use GuzzleHttp\Client;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 
 class RemoteConnector implements ConnectorInterface
@@ -14,6 +15,11 @@ class RemoteConnector implements ConnectorInterface
      */
     public function connect(array $config)
     {
-        return new RemoteQueue;
+        $client = new Client([
+            'base_uri' => $config['url'],
+            'auth' => [$config['username'], $config['password']],
+        ]);
+
+        return new RemoteQueue($client, $config['queue']);
     }
 }

src/RemoteQueue.php

@@ -2,12 +2,39 @@
 
 namespace Biigle\RemoteQueue;
 
+use GuzzleHttp\Client;
 use Illuminate\Queue\Queue;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 
 class RemoteQueue extends Queue implements QueueContract
 {
+    /**
+     * Guzzle client for the configured remote entpoint
+     *
+     * @var Client
+     */
+    protected $client;
+
+    /**
+     * Default queue to push jobs on the remote end
+     *
+     * @var string
+     */
+    protected $default;
+
+    /**
+     * Create a new remote queue instance.
+     *
+     * @param Client $client Guzzle client for the configured remote entpoint
+     * @param string $default Default queue to push jobs on the remote end
+     */
+    public function __construct(Client $client, $default = 'default')
+    {
+        $this->client = $client;
+        $this->default = $default;
+    }
+
     /**
      * Get the size of the queue.
      *
@@ -16,7 +43,10 @@ class RemoteQueue extends Queue implements QueueContract
      */
     public function size($queue = null)
     {
-        return 0;
+        $queue = $this->getQueue($queue);
+        $response = $this->client->get("queue/{$queue}/size");
+
+        return (int) $response->getBody()->getContents();
     }
 
     /**
@@ -29,7 +59,7 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        //
+        return $this->pushRaw($this->createPayload($job, $data), $queue);
     }
 
     /**
@@ -42,7 +72,9 @@ public function push($job, $data = '', $queue = null)
      */
     public function pushRaw($payload, $queue = null, array $options = [])
     {
-        //
+        $queue = $this->getQueue($queue);
+
+        return $this->client->post("queue/{$queue}", ['body' => $payload]);
     }
 
     /**
@@ -56,7 +88,8 @@ public function pushRaw($payload, $queue = null, array $options = [])
      */
     public function later($delay, $job, $data = '', $queue = null)
     {
-        //
+        // A delay is unsupported for now.
+        $this->pushRaw($this->createPayload($job, $data), $queue);
     }
 
     /**
@@ -69,4 +102,15 @@ public function pop($queue = null)
     {
         //
     }
+
+    /**
+     * Get the queue or return the default.
+     *
+     * @param  string|null  $queue
+     * @return string
+     */
+    protected function getQueue($queue)
+    {
+        return $queue ?: $this->default;
+    }
 }

tests/RemoteQueueTest.php

@@ -2,10 +2,94 @@
 
 namespace Biigle\RemoteQueue\Tests;
 
+use Queue;
+use GuzzleHttp\Client;
+use GuzzleHttp\Middleware;
+use GuzzleHttp\HandlerStack;
+use GuzzleHttp\Psr7\Response;
+use Illuminate\Bus\Queueable;
+use GuzzleHttp\Handler\MockHandler;
+use Biigle\RemoteQueue\RemoteQueue;
+
 class RemoteQueueTest extends TestCase
 {
-   public function testTest()
+   public function setUp()
+   {
+      parent::setUp();
+      config(['queue.connections.remote' => [
+         'driver' => 'remote',
+         'queue' => 'default',
+         'url' => 'http://localhost/api/remote-queue',
+         'username' => 'test',
+         'password' => 'test',
+      ]]);
+   }
+
+   public function testDriver()
+   {
+      $queue = Queue::connection('remote');
+      $this->assertInstanceOf(RemoteQueue::class, $queue);
+   }
+
+   public function testSize()
+   {
+      $mock = new MockHandler([new Response(200, [], 123)]);
+      $handler = HandlerStack::create($mock);
+      $container = [];
+      $handler->push(Middleware::history($container));
+      $client = new Client(['handler' => $handler]);
+      $queue = new RemoteQueue($client);
+
+      $size = $queue->size();
+      $this->assertEquals(123, $size);
+      $this->assertEquals('queue/default/size', $container[0]['request']->getUri());
+   }
+
+   public function testSizeQueue()
    {
-      //
+      $mock = new MockHandler([new Response(200, [], 123)]);
+      $handler = HandlerStack::create($mock);
+      $container = [];
+      $handler->push(Middleware::history($container));
+      $client = new Client(['handler' => $handler]);
+      $queue = new RemoteQueue($client);
+
+      $size = $queue->size('gpu');
+      $this->assertEquals(123, $size);
+      $this->assertEquals('queue/gpu/size', $container[0]['request']->getUri());
    }
+
+   public function testPush()
+   {
+      $mock = new MockHandler([new Response(200)]);
+      $handler = HandlerStack::create($mock);
+      $container = [];
+      $handler->push(Middleware::history($container));
+      $client = new Client(['handler' => $handler]);
+      $queue = new RemoteQueue($client);
+
+      $queue->push(new TestJob);
+      $request = $container[0]['request'];
+      $this->assertEquals('POST', $request->getMethod());
+      $this->assertEquals('queue/default', $request->getUri());
+      $this->assertContains('"commandName":"Biigle\\\RemoteQueue\\\Tests\\\TestJob"', $request->getBody()->getContents());
+   }
+
+   public function testPushQueue()
+   {
+      $mock = new MockHandler([new Response(200)]);
+      $handler = HandlerStack::create($mock);
+      $container = [];
+      $handler->push(Middleware::history($container));
+      $client = new Client(['handler' => $handler]);
+      $queue = new RemoteQueue($client);
+
+      $queue->push(new TestJob, '', 'gpu');
+      $this->assertEquals('queue/gpu', $container[0]['request']->getUri());
+   }
+}
+
+class TestJob
+{
+   use Queueable;
 }