feat: add websocket service

Author: normal-ai

composer.json

@@ -19,7 +19,8 @@
         "cathy\\AsyncTask\\Service"
       ],
       "config":{
-        "worker_task": "src/config/config.php"
+        "worker_task": "src/config/config.php",
+        "worker_websocket": "src/config/websocket.php"
       }
     }
   },

src/Service.php

@@ -6,6 +6,7 @@ class Service extends BaseService {
     public function register(){
         $this->commands([
             'worker:task'=>'\\cathy\\AsyncTask\\command\\AsyncTaskService',
+            'worker:websocket'=>'\\cathy\\AsyncTask\\command\\WebsocketService',
         ]);
     }
 }

src/command/WebsocketService.php

@@ -0,0 +1,90 @@
+<?php
+namespace cathy\AsyncTask\command;
+use think\console\Command;
+use think\console\Input;
+use think\console\input\Argument;
+use think\console\Output;
+use think\facade\Config;
+use Workerman\Worker;
+
+class WebsocketService extends Command
+{
+    protected $lastMtime;
+
+    public function configure()
+    {
+        $this->setName('worker:websocket')
+            ->addArgument('action', Argument::OPTIONAL, "start|stop|restart|reload|status|connections", 'start')
+            ->setDescription('Websocket Server for ThinkPHP6');
+    }
+
+    public function execute(Input $input, Output $output)
+    {
+        $configs = include_once __DIR__.'/../config/websocket.php';
+        $action = $input->getArgument('action');
+
+        if (DIRECTORY_SEPARATOR !== '\\') {
+            if (!in_array($action, ['start', 'stop', 'reload', 'restart', 'status', 'connections'])) {
+                $output->writeln("Invalid argument action:{$action}, Expected start|stop|restart|reload|status|connections .");
+                exit(1);
+            }
+
+            global $argv;
+            array_shift($argv);
+            array_shift($argv);
+            array_unshift($argv, 'think', $action);
+        }
+
+        if ('start' == $action) {
+            $output->writeln('Starting Websocket server...');
+        }
+
+        $option = Config::get('worker_websocket');
+
+        if ($input->hasOption('host')) {
+            $host = $input->getOption('host');
+        } else {
+            $host = !empty($option['host']) ? $option['host'] : $configs['host'];
+        }
+
+        if ($input->hasOption('port')) {
+            $port = $input->getOption('port');
+        } else {
+            $port = !empty($option['port']) ? $option['port'] : $configs['port'];
+        }
+
+        if($input->hasOption('daemon')){
+            $option['daemon'] = true;
+        } else {
+            $option['daemon'] = !empty($option['daemon']) ? $option['daemon'] : $configs['daemon'];
+        }
+
+        if(empty($option['name'])){
+            $option['name'] = $configs['name'];
+        }
+
+        if(empty($option['reuse_port'])){
+            $option['reuse_port'] = $configs['reuse_port'];
+        }
+
+        $this->start($host, (int) $port, $option);
+    }
+
+    public function start(string $host, int $port, array $option = []){
+        if(isset($option['daemon']) && $option['daemon']){
+            Worker::$daemonize = true;
+        }
+        $ws_worker = new Worker('websocket://'.$host.':'.$port);
+        $ws_worker->name = $option['name']; // 名称
+        $ws_worker->onConnect = function ($connection) {
+            echo 'New connection:'.$connection->getRemoteIp();
+        };
+        $ws_worker->onMessage = function($connection, $data) use ($option) {
+            $connection->send('Hello ' . $data);
+        };
+        $ws_worker->onClose = function ($connection) {
+            echo "Connection closed";
+        };
+        Worker::runAll();
+    }
+}

src/config/websocket.php

@@ -0,0 +1,8 @@
+<?php
+return [
+    'host'                  => '0.0.0.0',
+    'port'                  => 18080,
+    'name'                  => 'Websocket',
+    'daemon'                => true, // 是否开启守护进程
+    'reuse_port'            => false, // 是否开启端口复用
+];