Supported Swoole async queue driver

.github/ISSUE_TEMPLATE.md

@@ -1,4 +1,5 @@
-Please answer these questions before submitting your issue. Thanks!
+Make sure you read **Issues Guideline** and answer these questions before submitting your issue. Thanks!
+(Any **non-English** issues will be closed immediately.)
 
 1. Please provide your PHP and Swoole version. (`php -v` and `php --ri swoole`)
 

README.md

@@ -16,7 +16,7 @@ This package provides a high performance HTTP server to speed up your Laravel/Lu
 ## Features
 
 * Run **Laravel/Lumen** application on top of **Swoole**.
-* Outstanding performance boosting up to **30x**.
+* Outstanding performance boosting up to **5x** faster.
 * Sandbox mode to isolate app container.
 * Support running websocket server in **Laravel**.
 * Support `Socket.io` protocol.
@@ -28,7 +28,8 @@ Please see [Wiki](https://github.com/swooletw/laravel-swoole/wiki)
 
 ## Benchmark
 
-Test with clean Lumen 5.5, using MacBook Air 13, 2015.
+Test with clean Lumen 5.6, using DigitalOcean 3 CPUs / 1 GB Memory / PHP 7.2 / Ubuntu 16.04.4 x64
+
 Benchmarking Tool: [wrk](https://github.com/wg/wrk)
 
 ```
@@ -38,27 +39,31 @@ wrk -t4 -c100 http://your.app
 ### Nginx with FPM
 
 ```
-Running 10s test @ http://lumen.app:9999
-  4 threads and 100 connections
+wrk -t4 -c10 http://lumen-swoole.local
+
+Running 10s test @ http://lumen-swoole.local
+  4 threads and 10 connections
   Thread Stats   Avg      Stdev     Max   +/- Stdev
-    Latency     1.14s   191.03ms   1.40s    90.31%
-    Req/Sec    22.65     10.65    50.00     65.31%
-  815 requests in 10.07s, 223.65KB read
-Requests/sec:     80.93
-Transfer/sec:     22.21KB
+    Latency     6.41ms    1.56ms  19.71ms   71.32%
+    Req/Sec   312.99     28.71   373.00     72.00%
+  12469 requests in 10.01s, 3.14MB read
+Requests/sec:   1245.79
+Transfer/sec:    321.12KB
 ```
 
 ### Swoole HTTP Server
 
 ```
-Running 10s test @ http://127.0.0.1:1215
-  4 threads and 100 connections
+wrk -t4 -c10 http://lumen-swoole.local:1215
+
+Running 10s test @ http://lumen-swoole.local:1215
+  4 threads and 10 connections
   Thread Stats   Avg      Stdev     Max   +/- Stdev
-    Latency    11.58ms    4.74ms  68.73ms   81.63%
-    Req/Sec     2.19k   357.43     2.90k    69.50%
-  87879 requests in 10.08s, 15.67MB read
-Requests/sec:   8717.00
-Transfer/sec:      1.55MB
+    Latency     2.39ms    4.88ms 105.21ms   94.55%
+    Req/Sec     1.26k   197.13     1.85k    68.75%
+  50248 requests in 10.02s, 10.88MB read
+Requests/sec:   5016.94
+Transfer/sec:      1.09MB
 ```
 
 ## Q&A

src/HttpServiceProvider.php

@@ -9,6 +9,7 @@
 use SwooleTW\Http\Commands\HttpServerCommand;
 use Swoole\Websocket\Server as WebsocketServer;
 use SwooleTW\Http\Coroutine\Connectors\MySqlConnector;
+use SwooleTW\Http\Task\Connectors\SwooleTaskConnector;
 
 /**
  * @codeCoverageIgnore
@@ -45,10 +46,12 @@ public function register()
             $this->createSwooleServer();
             $this->configureSwooleServer();
         }
+
         $this->registerServer();
         $this->registerManager();
         $this->registerCommands();
         $this->registerDatabaseDriver();
+        $this->registerSwooleQueueDriver();
     }
 
     /**
@@ -133,9 +136,9 @@ protected function configureSwooleServer()
         $config = $this->app['config']->get('swoole_http.server.options');
 
         // only enable task worker in websocket mode
-        if (! $this->isWebsocket) {
-            unset($config['task_worker_num']);
-        }
+        // if (! $this->isWebsocket) {
+        //     unset($config['task_worker_num']);
+        // }
 
         static::$server->set($config);
     }
@@ -174,4 +177,16 @@ protected function registerDatabaseDriver()
             });
         });
     }
+
+    /**
+     * Register queue driver for swoole async task.
+     */
+    protected function registerSwooleQueueDriver()
+    {
+        $this->app->afterResolving('queue', function ($manager) {
+            $manager->addConnector('swoole', function () {
+                return new SwooleTaskConnector(static::$server);
+            });
+        });
+    }
 }

src/Server/Manager.php

@@ -5,6 +5,7 @@
 use Throwable;
 use Swoole\Http\Server;
 use SwooleTW\Http\Server\Sandbox;
+use SwooleTW\Http\Task\SwooleTaskJob;
 use Illuminate\Support\Facades\Facade;
 use SwooleTW\Http\Websocket\Websocket;
 use SwooleTW\Http\Transformers\Request;
@@ -228,13 +229,22 @@ public function onTask($server, $taskId, $srcWorkerId, $data)
         $this->container['events']->fire('swoole.task', func_get_args());
 
         try {
-            // push websocket message
-            if ($this->isWebsocket
-                && array_key_exists('action', $data)
-                && $data['action'] === Websocket::PUSH_ACTION) {
-                $this->pushMessage($server, $data['data'] ?? []);
+            if (is_array($data)) {
+                // push websocket message
+                if ($this->isWebsocket
+                    && array_key_exists('action', $data)
+                    && $data['action'] === Websocket::PUSH_ACTION) {
+                    $this->pushMessage($server, $data['data'] ?? []);
+                }
+            } elseif (is_string($data)) {
+                $decoded= json_decode($data, true);
+
+                if (JSON_ERROR_NONE === json_last_error() && isset($decoded['job'])) {
+                    (new SwooleTaskJob($this->container, $server, $data, $taskId, $srcWorkerId))->fire();
+                }
             }
         } catch (Throwable $e) {
+            dump($e);
             $this->logServerError($e);
         }
     }

src/Task/Connectors/SwooleTaskConnector.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace SwooleTW\Http\Task\Connectors;
+
+use SwooleTW\Http\Task\SwooleTaskQueue;
+use Illuminate\Queue\Connectors\ConnectorInterface;
+
+class SwooleTaskConnector implements ConnectorInterface
+{
+    /**
+     * Swoole Server Instance
+     *
+     * @var \Swoole\Http\Server
+     */
+    protected $swoole;
+
+     /**
+     * Create a new Swoole Async task connector instance.
+     *
+     * @param  \Swoole\Http\Server $swoole
+     * @return void
+     */
+    public function __construct($swoole)
+    {
+        $this->swoole = $swoole;
+    }
+
+    /**
+     * Establish a queue connection.
+     *
+     * @param  array  $config
+     * @return \Illuminate\Contracts\Queue\Queue
+     */
+    public function connect(array $config)
+    {
+        return new SwooleTaskQueue($this->swoole);
+    }
+}

src/Task/SwooleTaskJob.php

@@ -0,0 +1,83 @@
+<?php
+
+namespace SwooleTW\Http\Task;
+
+use Illuminate\Queue\Jobs\Job;
+use Illuminate\Contracts\Container\Container;
+use Illuminate\Contracts\Queue\Job as JobContract;
+
+class SwooleTaskJob extends Job implements JobContract
+{
+    /**
+     * The Swoole Server instance.
+     *
+     * @var \Swoole\Http\Server
+     */
+    protected $swoole;
+
+    /**
+     * The Swoole async job raw payload.
+     *
+     * @var array
+     */
+    protected $job;
+
+    /**
+     * The Task id
+     *
+     * @var int
+     */
+    protected $taskId;
+
+    /**
+     * The src worker Id
+     *
+     * @var int
+     */
+    protected $srcWrokerId;
+
+    /**
+     * Create a new job instance.
+     *
+     * @param  \Illuminate\Container\Container  $container
+     * @param  \Swoole\Http\Server  $swoole
+     * @param  string  $job
+     * @return void
+     */
+    public function __construct(Container $container, $swoole, $job, $taskId, $srcWrokerId)
+    {
+        $this->container = $container;
+        $this->swoole = $swoole;
+        $this->job = $job;
+        $this->taskId = $taskId;
+        $this->srcWorkderId = $srcWrokerId;
+    }
+
+    /**
+     * Get the number of times the job has been attempted.
+     * @return int
+     */
+    public function attempts()
+    {
+        return ($this->job['attempts'] ?? null) + 1;
+    }
+
+    /**
+     * Get the raw body string for the job.
+     * @return string
+     */
+    public function getRawBody()
+    {
+        return $this->job;
+    }
+
+
+    /**
+     * Get the job identifier.
+     * @return string
+     */
+    public function getJobId()
+    {
+        return $this->taskId;
+    }
+}

src/Task/SwooleTaskQueue.php

@@ -0,0 +1,105 @@
+<?php
+
+namespace SwooleTW\Http\Task;
+
+use Exception;
+use Illuminate\Queue\Queue;
+use Illuminate\Contracts\Queue\Queue as QueueContract;
+
+class SwooleTaskQueue extends Queue implements QueueContract
+{
+    /**
+     * Swoole Connector
+     *
+     * @var \Swoole\Http\Server
+     */
+    protected $swoole;
+
+    /**
+     * Create Async Task instance.
+     *
+     * @param \Swoole\Http\Server  $swoole
+     */
+    public function __construct($swoole)
+    {
+        $this->swoole = $swoole;
+    }
+
+
+    /**
+     * Push a new job onto the queue.
+     *
+     * @param  string|object  $job
+     * @param  mixed   $data
+     * @param  string  $queue
+     * @return mixed
+     */
+    public function push($job, $data = '', $queue = null)
+    {
+        return $this->pushRaw($this->createPayload($job, $data), $queue);
+    }
+
+    /**
+     * Push a raw payload onto the queue.
+     *
+     * @param  string  $payload
+     * @param  string  $queue
+     * @param  array   $options
+     * @return mixed
+     */
+    public function pushRaw($payload, $queue = null, array $options = [])
+    {
+        return $this->swoole->task($payload, ! is_numeric($queue) ? 1 : (int) $queue);
+    }
+
+    /**
+     * Push a new job onto the queue after a delay.
+     *
+     * @param  \DateTimeInterface|\DateInterval|int  $delay
+     * @param  string|object  $job
+     * @param  mixed   $data
+     * @param  string  $queue
+     * @return mixed
+     */
+    public function later($delay, $job, $data = '', $queue = null)
+    {
+        return swoole_timer_after($this->secondsUntil($delay), function () use ($job, $data, $queue) {
+            return $this->push($job, $data, $queue);
+        });
+    }
+
+    /**
+     * Create a typical, string based queue payload array.
+     *
+     * @param  string  $job
+     * @param  mixed  $data
+     *
+     * @throws Expcetion
+     */
+    protected function createStringPayload($job, $data)
+    {
+        throw new Exception("Unsupported empty data");
+    }
+
+    /**
+     * Get the size of the queue.
+     *
+     * @param  string  $queue
+     * @return int
+     */
+    public function size($queue = null)
+    {
+        return -1;
+    }
+
+    /**
+     * Pop the next job off of the queue.
+     *
+     * @param  string  $queue
+     * @return \Illuminate\Contracts\Queue\Job|null
+     */
+    public function pop($queue = null)
+    {
+        return null;
+    }
+}