diff --git a/ex/BASEServerListener.php b/ex/BASEServerListener.php new file mode 100644 index 00000000..0695a0c8 --- /dev/null +++ b/ex/BASEServerListener.php @@ -0,0 +1,227 @@ +server; + } + + + /** + * @param string $type + * @param string $host + * @param int $port + * @param int $mode + * @param array $settings + */ + public function addListener(string $type, string $host, int $port, int $mode, array $settings = []) + { + if (!$this->server) { + $this->createBaseServer($type, $host, $port, $mode, $settings); + } else { + if (!isset($settings['settings'])) { + $settings['settings'] = []; + } + $this->addNewListener($type, $host, $port, $mode, $settings); + } + } + + + /** + * startRun + */ + public function start(): void + { + $context = BASEServerListener::getContext(); + $configs = require_once 'server.php'; + foreach ($configs['servers']['handler'] as $config) { + $this->startListenerHandler($context, $config); + } + $context->server->start(); + } + + + /** + * @param BASEServerListener $context + * @param array $config + */ + private function startListenerHandler(BASEServerListener $context, array $config) + { + if ($this->server) { + $context->addNewListener($config['type'], $config['host'], $config['port'], $config['mode'], $config); + } else { + $config['settings'] = array_merge($configs['settings'] ?? [], $config['settings'] ?? []); + + $config['events'] = array_merge($configs['events'] ?? [], $config['events'] ?? []); + + $context->createBaseServer($config['type'], $config['host'], $config['port'], $config['mode'], $config); + } + } + + + /** + * @param string $type + * @param string $host + * @param int $port + * @param int $mode + * @param array $settings + */ + private function addNewListener(string $type, string $host, int $port, int $mode, array $settings = []) + { + $match = match ($type) { + self::SERVER_TYPE_TCP => TCPServerListener::class, + self::SERVER_TYPE_UDP => UDPServerListener::class, + self::SERVER_TYPE_HTTP => HTTPServerListener::class, + self::SERVER_TYPE_WEBSOCKET => WebSocketServerListener::class + }; + new $match($this->server, $host, $port, $mode, $settings); + } + + + /** + * @param string $type + * @param string $host + * @param int $port + * @param int $mode + * @param array $settings + */ + private function createBaseServer(string $type, string $host, int $port, int $mode, array $settings = []) + { + $match = match ($type) { + self::SERVER_TYPE_BASE, self::SERVER_TYPE_TCP, self::SERVER_TYPE_UDP => Server::class, + self::SERVER_TYPE_HTTP => HServer::class, + self::SERVER_TYPE_WEBSOCKET => WServer::class + }; + $this->server = new $match($host, $port, SWOOLE_PROCESS, $mode); + $this->server->set($settings['settings']); + $this->addDefaultListener($type, $settings); + } + + + /** + * @param string $type + * @param array $settings + * @return void + */ + private function addDefaultListener(string $type, array $settings): void + { + if ($this->server->setting['task_worker_num'] > 0) $this->addTaskListener($settings['events']); + if ($type === BASEServerListener::SERVER_TYPE_WEBSOCKET) { + $this->server->on('handshake', $settings['events'][static::SERVER_ON_HANDSHAKE] ?? [$this, 'nullHasNeed']); + $this->server->on('message', $settings['events'][static::SERVER_ON_MESSAGE] ?? [$this, 'nullHasNeed']); + $this->server->on('close', $settings['events'][static::SERVER_ON_CLOSE] ?? [$this, 'nullHasNeed']); + } else if ($type === BASEServerListener::SERVER_TYPE_UDP) { + $this->server->on('packet', $settings['events'][static::SERVER_ON_PACKET] ?? [$this, 'nullHasNeed']); + } else if ($type === BASEServerListener::SERVER_TYPE_HTTP) { + $this->server->on('request', $settings['events'][static::SERVER_ON_REQUEST] ?? [$this, 'nullHasNeed']); + } else { + $this->server->on('receive', $settings['events'][static::SERVER_ON_RECEIVE] ?? [$this, 'nullHasNeed']); + } + foreach ($settings['events'] as $event_type => $callback) { + if ($this->server->getCallback($event_type) !== null) { + continue; + } + $this->server->on($event_type, $callback); + } + } + + + /** + * @param array $events + */ + private function addTaskListener(array $events = []): void + { + $task_use_object = $this->server->setting['task_object'] ?? $this->server->setting['task_use_object'] ?? false; + if ($task_use_object || $this->server->setting['task_enable_coroutine']) { + $this->server->on('task', $events[static::SERVER_ON_TASK] ?? [ServerTask::class, 'onCoroutineTask']); + } else { + $this->server->on('task', $events[static::SERVER_ON_TASK] ?? [ServerTask::class, 'onTask']); + } + $this->server->on('finish', $events[static::SERVER_ON_FINISH] ?? [ServerTask::class, 'onFinish']); + } + + + public function nullHasNeed() + { + + } + +} + + +$context = BASEServerListener::getContext(); +$context->start(); diff --git a/ex/HTTPServerListener.php b/ex/HTTPServerListener.php new file mode 100644 index 00000000..f47d5868 --- /dev/null +++ b/ex/HTTPServerListener.php @@ -0,0 +1,65 @@ +_http = $server->addlistener($host, $port, $mode); + $this->_http->set($settings['settings'] ?? []); + if ($server->getCallback('request') === null) { + $server->on('request', $settings['events'][BASEServerListener::SERVER_ON_REQUEST] ?? [$this, 'onRequest']); + } + if (!in_array($server->setting['dispatch_mode'] ?? 2, [1, 3]) || $server->setting['enable_unsafe_event'] ?? false == true) { + $this->_http->on('connect', $settings['events'][BASEServerListener::SERVER_ON_CONNECT] ?? [$this, 'onConnect']); + $this->_http->on('close', $settings['events'][BASEServerListener::SERVER_ON_CLOSE] ?? [$this, 'onClose']); + } + } + + + /** + * @param Server $server + * @param int $fd + */ + public function onConnect(Server $server, int $fd) + { + var_dump(__FILE__ . ':' . __LINE__); + } + + + /** + * @param Request $request + * @param Response $response + */ + public function onRequest(Request $request, Response $response) + { + $response->setStatusCode(200); + $response->end(''); + } + + + /** + * @param Server $server + * @param int $fd + */ + public function onClose(Server $server, int $fd) + { + } + +} diff --git a/ex/SInterface/TaskExecute.php b/ex/SInterface/TaskExecute.php new file mode 100644 index 00000000..1a1233b7 --- /dev/null +++ b/ex/SInterface/TaskExecute.php @@ -0,0 +1,17 @@ +_tcp = $server->addlistener($host, $port, $mode); + $this->_tcp->set($settings); + $this->_tcp->on('receive', $settings['events'][BASEServerListener::SERVER_ON_RECEIVE] ?? [$this, 'onReceive']); + if (!in_array($server->setting['dispatch_mode'] ?? 2, [1, 3]) || $server->setting['enable_unsafe_event'] ?? false == true) { + $this->_tcp->on('connect', $settings['events'][BASEServerListener::SERVER_ON_CONNECT] ?? [$this, 'onConnect']); + $this->_tcp->on('close', $settings['events'][BASEServerListener::SERVER_ON_CLOSE] ?? [$this, 'onClose']); + } + } + + + /** + * @param Server $server + * @param int $fd + */ + public function onConnect(Server $server, int $fd) + { + var_dump(__FILE__ . ':' . __LINE__); + } + + + /** + * @param Server $server + * @param int $fd + * @param int $reactor_id + * @param string $data + */ + public function onReceive(Server $server, int $fd, int $reactor_id, string $data) + { + $server->send($fd, $data); + } + + + /** + * @param Server $server + * @param int $fd + */ + public function onClose(Server $server, int $fd) + { + } + +} diff --git a/ex/Task/ServerTask.php b/ex/Task/ServerTask.php new file mode 100644 index 00000000..b84293b9 --- /dev/null +++ b/ex/Task/ServerTask.php @@ -0,0 +1,70 @@ +execute(); + } catch (\Throwable $exception) { + $data = [$exception->getMessage()]; + } finally { + $server->finish(serialize($data)); + } + } + + + /** + * @param Server $server + * @param Server\Task $task + */ + public static function onCoroutineTask(Server $server, Server\Task $task) + { + try { + $data = unserialize($task->data); + if (!($data instanceof TaskExecute)) { + return; + } + $data->execute(); + } catch (\Throwable $exception) { + $data = [$exception->getMessage()]; + } finally { + $server->finish(serialize($data)); + } + } + + + /** + * @param Server $server + * @param int $task_id + * @param mixed $data + */ + public static function onFinish(Server $server, int $task_id, mixed $data) + { + if (!($data instanceof TaskExecute)) { + return; + } + $data->finish($server, $task_id); + } + + +} diff --git a/ex/UDPServerListener.php b/ex/UDPServerListener.php new file mode 100644 index 00000000..ab434e29 --- /dev/null +++ b/ex/UDPServerListener.php @@ -0,0 +1,43 @@ +_udp = $server->addlistener($host, $port, $mode); + $this->_udp->set($settings['settings'] ?? []); + $this->_udp->on('packet', $settings['events'][BASEServerListener::SERVER_ON_PACKET] ?? [$this, 'onPacket']); + } + + + /** + * @param Server $server + * @param string $data + * @param array $clientInfo + */ + public function onPacket(Server $server, string $data, array $clientInfo) + { + $server->sendto($clientInfo['address'], $clientInfo['port'], $data); + } + +} diff --git a/ex/WebSocketServerListener.php b/ex/WebSocketServerListener.php new file mode 100644 index 00000000..3d889c5b --- /dev/null +++ b/ex/WebSocketServerListener.php @@ -0,0 +1,75 @@ +_http = $server->addlistener($host, $port, $mode); + $this->_http->set($settings['settings'] ?? []); + $this->_http->on('handshake', $settings['events'][BASEServerListener::SERVER_ON_HANDSHAKE] ?? [$this, 'onHandshake']); + $this->_http->on('message', $settings['events'][BASEServerListener::SERVER_ON_MESSAGE] ?? [$this, 'onMessage']); + $this->_http->on('connect', $settings['events'][BASEServerListener::SERVER_ON_CONNECT] ?? [$this, 'onConnect']); + $this->_http->on('close', $settings['events'][BASEServerListener::SERVER_ON_CLOSE] ?? [$this, 'onClose']); + } + + + /** + * @param Request $request + * @param Response $response + */ + public function onHandshake(Request $request, Response $response) + { + + } + + + /** + * @param Server $server + * @param int $fd + */ + public function onConnect(Server $server, int $fd) + { + var_dump(__FILE__ . ':' . __LINE__); + } + + + /** + * @param \Swoole\WebSocket\Server|Server $server + * @param Frame $frame + */ + public function onMessage(\Swoole\WebSocket\Server|Server $server, Frame $frame) + { + } + + + /** + * @param Server $server + * @param int $fd + */ + public function onClose(Server $server, int $fd) + { + } + +} diff --git a/ex/server.php b/ex/server.php new file mode 100644 index 00000000..7b618605 --- /dev/null +++ b/ex/server.php @@ -0,0 +1,95 @@ + [ + 'settings' => [ + 'worker_num' => swoole_cpu_num() * 3, + 'reactor_num' => swoole_cpu_num(), + 'log_file' => APP_PATH . 'storage/request.log', + 'stats_file' => APP_PATH . 'storage/stats.log', + 'dispatch_mode' => 3, + 'task_worker_num' => 1, + 'enable_coroutine' => true, + 'task_enable_coroutine' => true, + 'daemonize' => 0, + 'open_tcp_keepalive' => 1, + 'heartbeat_check_interval' => 60, + 'heartbeat_idle_time' => 600, + 'tcp_keepidle' => 3, + 'tcp_keepinterval' => 1, + 'tcp_keepcount' => 2, + 'max_wait_time' => 60, + 'reload_async' => true, + 'tcp_fastopen' => 1, + 'tcp_defer_accept' => 1 + ], + 'events' => [ + BASEServerListener::SERVER_ON_PIPE_MESSAGE => [], + BASEServerListener::SERVER_ON_SHUTDOWN => [], + BASEServerListener::SERVER_ON_TASK => [], + BASEServerListener::SERVER_ON_WORKER_START => [], + BASEServerListener::SERVER_ON_WORKER_ERROR => [], + BASEServerListener::SERVER_ON_WORKER_EXIT => [], + BASEServerListener::SERVER_ON_WORKER_STOP => [], + BASEServerListener::SERVER_ON_MANAGER_START => [], + BASEServerListener::SERVER_ON_MANAGER_STOP => [], + BASEServerListener::SERVER_ON_BEFORE_RELOAD => [], + BASEServerListener::SERVER_ON_AFTER_RELOAD => [], + BASEServerListener::SERVER_ON_START => [], + ], + 'handler' => [ + [ + 'type' => BASEServerListener::SERVER_TYPE_WEBSOCKET, + 'host' => '0.0.0.0', + 'port' => 9001, + 'mode' => SWOOLE_SOCK_TCP, + 'settings' => [ + 'open_http_protocol' => false, + 'open_http2_protocol' => false + ], + 'events' => [ + BASEServerListener::SERVER_ON_CONNECT => [WebSocketServerListener::class, 'onConnect'], + BASEServerListener::SERVER_ON_HANDSHAKE => [WebSocketServerListener::class, 'onHandshake'], + BASEServerListener::SERVER_ON_MESSAGE => [WebSocketServerListener::class, 'onMessage'], + BASEServerListener::SERVER_ON_CLOSE => [WebSocketServerListener::class, 'onClose'], + ] + ], [ + 'type' => BASEServerListener::SERVER_TYPE_HTTP, + 'host' => '0.0.0.0', + 'port' => 9001, + 'mode' => SWOOLE_SOCK_TCP, + 'events' => [ + BASEServerListener::SERVER_ON_REQUEST => [HTTPServerListener::class, 'onRequest'], + ], + 'settings' => [ + 'open_http_protocol' => true, + 'open_http2_protocol' => false, + 'upload_tmp_dir' => APP_PATH . 'storage', + 'http_parse_cookie' => true, + 'http_compression' => true, + 'http_compression_level' => 5, + 'enable_unsafe_event' => false, + ] + ], [ + 'type' => BASEServerListener::SERVER_TYPE_TCP, + 'host' => '0.0.0.0', + 'port' => 9001, + 'mode' => SWOOLE_SOCK_TCP, + 'events' => [ + BASEServerListener::SERVER_ON_CONNECT => [TCPServerListener::class, 'onConnect'], + BASEServerListener::SERVER_ON_RECEIVE => [TCPServerListener::class, 'onReceive'], + BASEServerListener::SERVER_ON_CLOSE => [TCPServerListener::class, 'onClose'], + ] + ], [ + 'type' => BASEServerListener::SERVER_TYPE_UDP, + 'host' => '0.0.0.0', + 'port' => 9001, + 'mode' => SWOOLE_SOCK_TCP, + 'events' => [ + BASEServerListener::SERVER_ON_PACKET => [UDPServerListener::class, 'onPacket'], + ] + ], + ] + ] +]; diff --git a/ex/test/RegisterTask.php b/ex/test/RegisterTask.php new file mode 100644 index 00000000..3aa0d742 --- /dev/null +++ b/ex/test/RegisterTask.php @@ -0,0 +1,41 @@ +