Implement QueueController to receive jobs

Author: mzur

README.md

@@ -1,6 +1,6 @@
 # Remote Queue
 
-Submit jobs to other Laravel or Lumen instances.
+Submit jobs to and receive jobs from other Laravel or Lumen instances.
 
 ## Installation
 
@@ -19,11 +19,49 @@ Add `$app->register(Biigle\RemoteQueue\RemoteQueueServiceProvider::class);` to `
 
 ## Configuration
 
-todo
+You can override any of these configuration options of the `remote-queue` config either directly or via environment variables:
+
+### remote-queue.listen
+
+Default: `false`
+Environment: `REMOTE_QUEUE_LISTEN`
+
+Accept and process jobs sent to this application instance.
+
+### remote-queue.endpoint
+
+Default: `api/v1/remote-queue`
+Environment: `REMOTE_QUEUE_ENDPOINT`
+
+API endpoint to receive new jobs.
+
+### remote-queue.connection
+
+Default: `null`
+Environment: `REMOTE_QUEUE_CONNECTION`
+
+Use this queue connection to process received jobs. If `null`, the default connection is used.
+
+### remote-queue.accept-tokens
+
+Default: `[]`
+Environment: `REMOTE_QUEUE_ACCEPT_TOKENS`
+
+Accept jobs only if they provide one of these tokens. Specify tokens as comma separated list if you use the environment variable. One way to generate tokens is this command: `head -c 32 /dev/urandom | base64`.
 
 ## Usage
 
-To use the remote queue, configure a queue connection in the `queue.connections` config to use the `remote` driver:
+This package can be used to submit queued jobs to another Laravel or Lumen application, receive jobs from another application or both.
+
+### Receive jobs
+
+By default, this package does not allow receiving of jobs from another application. To allow this, set the `remote-queue.listen` config to `true`. Tokens are used to authenticate incoming requests. A token is just some long random string in this case. Configure `remote-queue.accept-tokens` with all tokens that are accepted. All received jobs which are successfully authenticated are pushed to a "regular" queue of this application and processed by the queue worker.
+
+**Important:** Make sure that the job class exists in both the submitting and receiving application.
+
+### Submit jobs
+
+Jobs pushed to the remote queue are transmitted via HTTP and processed on another application. To use the remote queue, configure a queue connection in the `queue.connections` config to use the `remote`. Example:
 
 ```php
 [
@@ -32,20 +70,20 @@ To use the remote queue, configure a queue connection in the `queue.connections`
    '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' => '',
+   // Token to use for authentication on the remote host.
+   'token' => 'IoO2l7UKZfso5zQloF2XvAShEbAR5a9M8u+WBfg0HgI=',
 ]
 ```
 
+You can now dispatch jobs to the remote queue connection just like any other connection.
+
 ```php
 use App\Jobs\MyJob;
 
 MyJob::dispatch($data)->onConnection('remote');
 ```
 
-## Submit/Response Pattern
+### Submit/Response Pattern
 
 We developed this package to be able to process jobs on a remote machine with GPU. To return the computed results, we applied what we call the "submit/response" pattern.
 

composer.json

@@ -12,7 +12,8 @@
         "guzzlehttp/guzzle": "~6.0",
         "illuminate/queue": "^5.5",
         "illuminate/support": "^5.5",
-        "illuminate/contracts": "^5.5"
+        "illuminate/contracts": "^5.5",
+        "illuminate/routing": "^5.5"
     },
     "autoload": {
         "psr-4": {
@@ -36,6 +37,7 @@
     },
     "require-dev": {
         "laravel/laravel": "^5.5",
-        "phpunit/phpunit": "^7.4"
+        "phpunit/phpunit": "^7.4",
+        "mockery/mockery": "^1.2"
     }
 }

composer.lock

@@ -18,4 +18,11 @@
             <directory suffix=".php">src</directory>
         </whitelist>
     </filter>
+
+    <php>
+        <env name="APP_DEBUG" value="true"/>
+        <env name="APP_ENV" value="testing"/>
+        <env name="CACHE_DRIVER" value="array"/>
+        <env name="REMOTE_QUEUE_LISTEN" value="true"/>
+    </php>
 </phpunit>

phpunit.xml

@@ -0,0 +1,44 @@
+<?php
+
+namespace Biigle\RemoteQueue\Http\Controllers;
+
+use Illuminate\Http\Request;
+use Illuminate\Http\Response;
+use Illuminate\Routing\Controller;
+use Illuminate\Queue\QueueManager;
+
+class QueueController extends Controller
+{
+    /**
+     * Get the size of a queue
+     *
+     * @param QueueManager $manager
+     * @param string $queue
+     *
+     * @return int
+     */
+    public function show(QueueManager $manager, $queue)
+    {
+        return $manager->connection()->size($queue);
+    }
+
+    /**
+     * Accept a new job and push it to the local queue.
+     *
+     * @param Request $request
+     * @param QueueManager $manager
+     * @param string $queue
+     */
+    public function store(Request $request, QueueManager $manager, $queue)
+    {
+        $payload = $request->getContent();
+
+        if (!$payload) {
+            return new Response('Job payload is required', 422);
+        }
+
+        $manager->connection(config('remote-queue.connection'))
+            ->pushRaw($payload, $queue);
+
+    }
+}

src/Http/Controllers/QueueController.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Biigle\RemoteQueue\Http\Middleware;
+
+use Closure;
+
+class Authenticate
+{
+    /**
+     * Handle an incoming request.
+     *
+     * @param  \Illuminate\Http\Request  $request
+     * @param  \Closure  $next
+     * @param  string|null  $guard
+     * @return mixed
+     */
+    public function handle($request, Closure $next, $guard = null)
+    {
+        $token = $request->bearerToken();
+        if ($token && in_array($token, config('remote-queue.accept-tokens'))) {
+            return $next($request);
+        }
+
+        return response('Unauthorized.', 401);
+    }
+}

src/Http/Middleware/Authenticate.php

@@ -17,7 +17,7 @@ public function connect(array $config)
     {
         $client = new Client([
             'base_uri' => $config['url'],
-            'auth' => [$config['username'], $config['password']],
+            'headers' => ['Authorization' => "Bearer {$config['token']}"]
         ]);
 
         return new RemoteQueue($client, $config['queue']);

src/RemoteConnector.php

@@ -44,7 +44,7 @@ public function __construct(Client $client, $default = 'default')
     public function size($queue = null)
     {
         $queue = $this->getQueue($queue);
-        $response = $this->client->get("queue/{$queue}/size");
+        $response = $this->client->get("{$queue}/size");
 
         return (int) $response->getBody()->getContents();
     }
@@ -74,7 +74,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
     {
         $queue = $this->getQueue($queue);
 
-        return $this->client->post("queue/{$queue}", ['body' => $payload]);
+        return $this->client->post("{$queue}", ['body' => $payload]);
     }
 
     /**

src/RemoteQueue.php

@@ -2,8 +2,10 @@
 
 namespace Biigle\RemoteQueue;
 
+use Illuminate\Routing\Router;
 use Illuminate\Queue\QueueManager;
 use Illuminate\Support\ServiceProvider;
+use Biigle\RemoteQueue\Http\Middleware\Authenticate;
 
 class RemoteQueueServiceProvider extends ServiceProvider
 {
@@ -12,11 +14,23 @@ class RemoteQueueServiceProvider extends ServiceProvider
      *
      * @return void
      */
-    public function boot(QueueManager $manager)
+    public function boot(QueueManager $manager, Router $router)
     {
         $manager->addConnector('remote', function() {
             return new RemoteConnector;
         });
+
+        if (config('remote-queue.listen')) {
+            $router->group([
+                'prefix' => config('remote-queue.endpoint'),
+                'namespace' => 'Biigle\RemoteQueue\Http\Controllers',
+                'middleware' => Authenticate::class,
+            ], function ($router) {
+                $router->post('{queue}', 'QueueController@store');
+                $router->get('{queue}/size', 'QueueController@show');
+            });
+        }
+
     }
 
     /**
@@ -26,6 +40,6 @@ public function boot(QueueManager $manager)
      */
     public function register()
     {
-        //
+        $this->mergeConfigFrom(__DIR__.'/config/remote-queue.php', 'remote-queue');
     }
 }