变更

2023-04-22 02:04:31 +08:00
parent 3c0581c2f6
commit ba81504942
9 changed files with 446 additions and 401 deletions
@@ -5,21 +5,21 @@ namespace Kiri\Server\Abstracts;
 use Exception;
 use Kiri;
 use Kiri\Abstracts\Config;
-use Kiri\Server\Handler\OnServer;
+use Kiri\Abstracts\Logger;
 use Kiri\Exception\ConfigException;
+use Kiri\Exception\NotFindClassException;
+use Kiri\Server\Config as SConfig;
+use Kiri\Server\Constant;
+use Kiri\Server\Events\OnServerBeforeStart;
 use Kiri\Server\Events\OnShutdown;
+use Kiri\Server\Handler\OnServer;
+use Kiri\Server\ServerInterface;
+use Kiri\Server\Task\TaskInterface;
+use Kiri\Server\Task\Task;
 use Psr\Container\ContainerExceptionInterface;
 use Psr\Container\NotFoundExceptionInterface;
 use ReflectionException;
-use Kiri\Server\Config as SConfig;
-use Kiri\Di\LocalService;
 use Swoole\Server;
-use Kiri\Server\ServerInterface;
-use Kiri\Server\Constant;
-use Kiri\Events\EventDispatch;
-use Kiri\Exception\NotFindClassException;
-use Kiri\Server\Events\OnServerBeforeStart;
-use Kiri\Abstracts\Logger;

 /**
 *
@@ -33,46 +33,63 @@ class AsyncServer implements ServerInterface
 	/**
 	 * @var Server|null
 	 */
-	private Server|null $server = null;
+	private ?Server $server = null;


 	/**
 	 * @param array $service
 	 * @param int $daemon
 	 * @return void
-	 * @throws ConfigException
-	 * @throws ContainerExceptionInterface
-	 * @throws NotFindClassException
-	 * @throws NotFoundExceptionInterface
 	 * @throws Exception
 	 */
 	public function initCoreServers(array $service, int $daemon = 0): void
 	{
-		$service = $this->genConfigService($service);
-
-		$this->createBaseServer(array_shift($service), $daemon);
-		foreach ($service as $value) {
-			$this->addListener($value);
-		}
-		$rpcService = Config::get('rpc', []);
-		if (!empty($rpcService)) {
-			$this->addListener(instance(SConfig::class, [], $rpcService));
-		}
-
-		$processManager = Kiri::getDi()->get(ProcessManager::class);
-		$processManager->batch(Config::get('processes', []));
-
-		$this->onSignal(Config::get('signal', []));
+		$this->listener($service, $daemon);
+		$this->initRpcListen();
+		$this->initProcess();
+		$this->onSignal();
 	}


 	/**
-	 * @param string $name
-	 * @return Server|null
+	 * @param array $service
+	 * @param $daemon
+	 * @return void
+	 * @throws
 	 */
-	public function getServer(string $name = ''): Server|null
+	private function listener(array $service, $daemon): void
 	{
-		return $this->server;
+		$service = $this->genConfigService($service);
+		foreach ($service as $value) {
+			if (is_null($this->server)) {
+				$this->createBaseServer($value, $daemon);
+			} else {
+				$this->addListener($value);
+			}
+		}
+	}
+
+
+	/**
+	 * @return void
+	 * @throws Exception
+	 */
+	private function initRpcListen(): void
+	{
+		$rpcService = Config::get('rpc', []);
+		if (!empty($rpcService)) {
+			$this->addListener(instance(SConfig::class, [], $rpcService));
+		}
+	}
+
+	/**
+	 * @return void
+	 */
+	private function initProcess(): void
+	{
+		foreach ($this->_process as $process) {
+			$this->server->addProcess($process);
+		}
 	}


@@ -85,8 +102,7 @@ class AsyncServer implements ServerInterface
 	{
 		$this->server->shutdown();

-		$processManager = Kiri::getDi()->get(EventDispatch::class);
-		$processManager->dispatch(new OnShutdown());
+		event(new OnShutdown());

 		return true;
 	}
@@ -106,6 +122,23 @@ class AsyncServer implements ServerInterface
 		if (is_null($match)) {
 			throw new NotFindClassException('Unknown server type ' . $config->type);
 		}
+		$this->initServer($match, $config, $daemon);
+		$this->onEventListen($this->server, Config::get('server.events', []));
+		$this->onEventListen($this->server, $config->events);
+		$this->onTaskListen();
+	}
+
+
+	/**
+	 * @param $match
+	 * @param $config
+	 * @param $daemon
+	 * @return void
+	 * @throws ConfigException
+	 * @throws ReflectionException
+	 */
+	private function initServer($match, $config, $daemon): void
+	{
 		$this->server = new $match($config->host, $config->port, $config->mode, $config->socket);
 		$this->server->set($this->systemConfig($config, $daemon));

@@ -113,8 +146,23 @@ class AsyncServer implements ServerInterface
 		if (!isset($config->events[Constant::SHUTDOWN])) {
 			$config->events[Constant::SHUTDOWN] = [OnServer::class, 'onShutdown'];
 		}
-		$this->onEventListen($this->server, Config::get('server.events', []));
-		$this->onEventListen($this->server, $config->events);
+		Kiri::service()->set('server', $this->server);
+	}
+
+
+	/**
+	 * @return void
+	 * @throws
+	 */
+	private function onTaskListen(): void
+	{
+		if (!isset($this->server->setting[Constant::OPTION_TASK_WORKER_NUM])) {
+			return;
+		}
+		$container = Kiri::getDi();
+		$task = $container->get(Task::class);
+		$container->set(TaskInterface::class, $task);
+		$task->initTaskWorker($this->server);
 	}


@@ -150,12 +198,11 @@ class AsyncServer implements ServerInterface
 			throw new Exception('Listen port fail.' . swoole_last_error());
 		}

-		\Kiri::getLogger()->alert('Listen ' . $config->type . ' address ' . $config->host . '::' . $config->port);
+		Logger::_waring('Listen ' . $config->type . ' address ' . $config->host . '::' . $config->port);

 		$port->set($this->resetSettings($config->type, $config->settings));

 		$this->onEventListen($port, $config->getEvents());
-		Kiri::getDi()->get(LocalService::class)->set($config->getName(), $port);
 	}

 	/**
@@ -167,7 +214,7 @@ class AsyncServer implements ServerInterface
 	public function onSigint($no, array $signInfo): void
 	{
 		try {
-			\Kiri::getLogger()->alert('Pid ' . getmypid() . ' get signo ' . $no);
+			Logger::_alert('Pid ' . getmypid() . ' get signo ' . $no);
 			$this->shutdown();
 		} catch (\Throwable $exception) {
 			error($exception);
@@ -179,6 +226,7 @@ class AsyncServer implements ServerInterface
 	 * @param string $type
 	 * @param array $settings
 	 * @return array
+	 * @throws
 	 */
 	private function resetSettings(string $type, array $settings): array
 	{
@@ -3,12 +3,11 @@
 namespace Kiri\Server\Abstracts;

 use Exception;
-use Kiri\Server\CoroutineServer;
-use Psr\Container\ContainerExceptionInterface;
-use Psr\Container\NotFoundExceptionInterface;
+use Kiri;
 use ReflectionException;
 use Swoole\Coroutine;
 use Swoole\Http\Server as HServer;
+use Swoole\Process;
 use Swoole\Server;
 use Kiri\Server\Constant;
 use Kiri\Server\Config;
@@ -16,11 +15,11 @@ use Swoole\WebSocket\Server as WServer;

 trait TraitServer
 {
-	
-	
+
+
 	private array $_process = [];
-	
-	
+
+
 	/**
 	 * @param string|array|BaseProcess $class
 	 * @return void
@@ -28,30 +27,48 @@ trait TraitServer
 	 */
 	public function addProcess(string|array|BaseProcess $class): void
 	{
-		$container = \Kiri::getDi()->get(ProcessManager::class);
-		
 		if (!is_array($class)) {
 			$class = [$class];
 		}
 		foreach ($class as $name) {
-			$container->add($name);
+			if (is_string($name)) {
+				$name = Kiri::getDi()->get($name);
+			}
+			if (isset($this->_process[$name->getName()])) {
+				throw new Exception('Process(' . $name->getName() . ') is exists.');
+			}
+			$this->_process[$name->getName()] = $this->genProcess($name);
 		}
 	}
-	
-	
+
+
+	/**
+	 * @param BaseProcess $name
+	 * @return Process
+	 */
+	private function genProcess(BaseProcess $name): Process
+	{
+		return new Process(function (Process $process) use ($name) {
+			$process->name($name->getName());
+			$name->onSigterm()->process($process);
+		},
+			$name->getRedirectStdinAndStdout(),
+			$name->getPipeType(),
+			$name->isEnableCoroutine());
+	}
+
+
 	/**
-	 * @param array $signal
 	 * @return void
-	 * @throws ContainerExceptionInterface
-	 * @throws NotFoundExceptionInterface
 	 * @throws Exception
 	 */
-	public function onSignal(array $signal): void
+	public function onSignal(): void
 	{
+		$signal = \Kiri\Abstracts\Config::get('signal', []);
 		$this->onPcntlSignal(SIGINT, [$this, 'onSigint']);
 		foreach ($signal as $sig => $value) {
 			if (is_array($value) && is_string($value[0])) {
-				$value[0] = $this->container->get($value[0]);
+				$value[0] = \Kiri::getDi()->get($value[0]);
 			}
 			if (!is_callable($value, true)) {
 				throw new Exception('Register signal callback must can exec.');
@@ -59,8 +76,8 @@ trait TraitServer
 			$this->onPcntlSignal($sig, $value);
 		}
 	}
-	
-	
+
+
 	/**
 	 * @param $signal
 	 * @param $callback
@@ -68,19 +85,19 @@ trait TraitServer
 	 */
 	private function onPcntlSignal($signal, $callback): void
 	{
-		if (get_called_class() != CoroutineServer::class) {
-			pcntl_signal(SIGINT, [$this, 'onSigint']);
-		} else {
-			Coroutine::create(static function () use ($signal, $callback) {
-				$data = Coroutine::waitSignal($signal);
-				if ($data) {
-					$callback($signal, [true]);
-				}
-			});
-		}
+//		if (get_called_class() != CoroutineServer::class) {
+		pcntl_signal(SIGINT, [$this, 'onSigint']);
+//		} else {
+//			Coroutine::create(static function () use ($signal, $callback) {
+//				$data = Coroutine::waitSignal($signal);
+//				if ($data) {
+//					$callback($signal, [true]);
+//				}
+//			});
+//		}
 	}
-	
-	
+
+
 	/**
 	 * @return array
 	 */
@@ -88,8 +105,8 @@ trait TraitServer
 	{
 		return $this->_process;
 	}
-	
-	
+
+
 	/**
 	 * @param array $ports
 	 * @return array
@@ -139,8 +156,8 @@ trait TraitServer
 		}
 		return $array;
 	}
-	
-	
+
+
 	/**
 	 * @param $type
 	 * @return string|null
@@ -155,8 +172,8 @@ trait TraitServer
 			default => null
 		};
 	}
-	
-	
+
+
 	/**
 	 * @param $type
 	 * @return string|null
@@ -169,6 +186,6 @@ trait TraitServer
 			default => null
 		};
 	}
-	
-	
+
+
 }
@@ -0,0 +1,166 @@
+<?php
+
+namespace Kiri\Server\Abstracts;
+
+use Exception;
+use Kiri\Abstracts\Component;
+use Kiri\Router\DataGrip;
+use Kiri\Router\Handler;
+use Kiri\Router\RouterCollector;
+use Kiri\Server\Contract\OnCloseInterface;
+use Kiri\Server\Contract\OnDisconnectInterface;
+use Kiri\Server\Contract\OnHandshakeInterface;
+use Kiri\Server\Contract\OnMessageInterface;
+use ReflectionException;
+use Swoole\Http\Request;
+use Swoole\Http\Response;
+use Swoole\Server;
+use Swoole\WebSocket\Frame;
+
+
+class WebSocketServer extends Component implements OnHandshakeInterface, OnMessageInterface, OnDisconnectInterface, OnCloseInterface
+{
+
+
+	public string $host = '127.0.0.1';
+
+
+	public int $port = 6086;
+
+
+	public array $events = [];
+
+	public array $settings = [];
+
+	public int $socket_type = SWOOLE_SOCK_TCP6;
+
+
+	public Handler $handler;
+
+
+	/**
+	 * @var RouterCollector
+	 */
+	public RouterCollector $collector;
+
+
+	/**
+	 * @return void
+	 * @throws ReflectionException
+	 */
+	public function init(): void
+	{
+		/** @var Server $server */
+		$application = \Kiri::service();
+		if (!$application->has('server')) {
+			$server = new \Swoole\WebSocket\Server($this->host, $this->port, SWOOLE_PROCESS);
+			$application->set('server', $server);
+		} else {
+			$server = $application->get('server');
+		}
+
+		$socket = $server->addlistener($this->host, $this->port, $this->socket_type);
+		$socket->set($this->settings);
+		$socket->on('handshake', [$this, 'onHandshake']);
+		$socket->on('message', [$this, 'onMessage']);
+		$socket->on('disconnect', [$this, 'onDisconnect']);
+
+		$this->collector = \Kiri::getDi()->get(DataGrip::class)->get('wss');
+
+		$this->handler = $this->collector->query('/', 'GET');
+	}
+
+
+	/**
+	 * @param Request $request
+	 * @param Response $response
+	 * @return void
+	 * @throws Exception
+	 */
+	public function onHandshake(Request $request, Response $response): void
+	{
+		// TODO: Implement onHandshake() method.
+		$secWebSocketKey = $request->header['sec-websocket-key'];
+		$patten = '#^[+/0-9A-Za-z]{21}[AQgw]==$#';
+		if (0 === preg_match($patten, $secWebSocketKey) || 16 !== strlen(base64_decode($secWebSocketKey))) {
+			throw new Exception('protocol error.', 500);
+		}
+		$key = base64_encode(sha1($request->header['sec-websocket-key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', TRUE));
+		$headers = [
+			'Upgrade'               => 'websocket',
+			'Connection'            => 'Upgrade',
+			'Sec-websocket-Accept'  => $key,
+			'Sec-websocket-Version' => '13',
+		];
+		if (isset($request->header['sec-websocket-protocol'])) {
+			$explode = explode(',', $request->header['sec-websocket-protocol']);
+			$headers['Sec-websocket-Protocol'] = $explode[0];
+		}
+		foreach ($headers as $key => $val) {
+			$response->setHeader($key, $val);
+		}
+		if ($this->handler->implement(OnHandshakeInterface::class)) {
+			$handler = $this->handler->getClass();
+
+			call_user_func([$handler, 'onHandshake'], $request, $response);
+		} else {
+			$response->setStatusCode(101);
+			$response->end();
+		}
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param Frame $frame
+	 * @return void
+	 */
+	public function onMessage(Server $server, Frame $frame): void
+	{
+		// TODO: Implement onMessage() method.
+		if (!$this->handler->implement(OnMessageInterface::class)) {
+			return;
+		}
+
+		$handler = $this->handler->getClass();
+
+		call_user_func([$handler, 'onMessage'], $server, $frame);
+	}
+
+
+	/**
+	 * @param \Swoole\WebSocket\Server $server
+	 * @param int $fd
+	 * @return void
+	 */
+	public function onDisconnect(\Swoole\WebSocket\Server $server, int $fd): void
+	{
+		// TODO: Implement onDisconnect() method.
+		if (!$this->handler->implement(OnDisconnectInterface::class)) {
+			return;
+		}
+
+		$handler = $this->handler->getClass();
+
+		call_user_func([$handler, 'onDisconnect'], $server, $fd);
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param int $fd
+	 * @return void
+	 */
+	public function onClose(Server $server, int $fd): void
+	{
+		// TODO: Implement onDisconnect() method.
+		if (!$this->handler->implement(OnCloseInterface::class)) {
+			return;
+		}
+
+		$handler = $this->handler->getClass();
+
+		call_user_func([$handler, 'onClose'], $server, $fd);
+	}
+
+}