server = new Coroutine\Http\Server($this->host, $this->port, $this->isSsl, $this->reuse_port); $this->server->set(['max_coroutine' => 500000]); $this->server->handle('/', [$this, 'actor']); } /** * @param Coroutine\WaitGroup $group * @return void */ public function run(Coroutine\WaitGroup $group): void { Coroutine::create(function () use ($group) { $this->dispatch->dispatch(new OnWorkerStart(null, 0)); $this->start($group); }); } /** * @param WaitGroup $group * @return void * @throws ContainerExceptionInterface * @throws NotFoundExceptionInterface * @throws ReflectionException */ public function start(WaitGroup $group): void { $this->server->start(); $this->dispatch->dispatch(new OnBeforeShutdown()); if ($this->isShutdown === false) { $this->start($group); } else { $group->done(); } } public function stop() { $this->isShutdown = true; $this->server->shutdown(); } /** * @param $path * @param $method * @param array|Closure $closure * @return void */ public function handler($path, $method, array|Closure $closure): void { $this->router[$path] = [strtolower($method), $closure]; } /** * @param Request $request * @param Response $response * @return mixed * @throws Exception */ public function actor(Request $request, Response $response): mixed { Context::setContext(ResponseInterface::class, new \Kiri\Message\Response()); Context::setContext(RequestInterface::class, ServerRequest::createServerRequest($request)); if ($request['request_method'] === 'HEAD') { return $this->write('', $response, 200); } [$method, $handler] = $this->router[$request['request_uri']] ?? [null, null]; if (is_null($handler)) { return $this->write('Page not found.', $response, 404); } if (!is_callable($handler, true)) { return $this->write('Page not found.', $response, 404); } if ($method !== $request['request_method']) { return $this->write('Page allow.', $response, 405); } if (isset($request->header['upgrade']) && $request->header['upgrade'] == 'websocket') { defer(function () use ($handler, $request) { if (!$handler instanceof OnOpenInterface) { return; } $handler->onOpen($this->server, $request); }); if ($handler instanceof OnHandshakeInterface) { $handler->onHandshake($request, $response); } else { $response->upgrade(); } while (true) { $read = $response->recv(); if ($read === '' || $read === null || $read instanceof CloseFrame) { break; } if ($handler instanceof OnMessageInterface) { $handler->onMessage($this->server, $read); } } if ($handler instanceof OnCloseInterface) { $handler->onClose($response->fd); } return null; } $params = $this->container->getArgs($handler[1], $handler[0] ?? null); $result = call_user_func($handler, ...$params); if (is_null($result)) { return $this->write("", $response); } else { return $this->write($result, $response); } } /** * @param mixed $message * @param Response $response * @param int $statusCode * @return mixed */ private function write(mixed $message, Response $response, int $statusCode = 200): mixed { $result = $message; if ($message instanceof ResponseInterface) { $result = $result->getBody()->getContents(); $response->setStatusCode($message->getStatusCode()); } else { $message = Context::getContext(ResponseInterface::class); $response->setStatusCode($statusCode); } $headers = $message->getHeaders(); if (is_array($headers)) foreach ($headers as $key => $header) { $response->setHeader($key, $header); } if (!isset($response->header['content-type'])) { $response->header('content-type', 'application/json'); } else if (!isset($response->header['Content-Type'])) { $response->header('content-type', 'application/json'); } $headers = $message->getCookieParams(); if (is_array($headers)) foreach ($headers as $header) { $response->cookie($header->name, $header->value, $header->expires, $header->path, $header->domain, $header->secure, $header->httponly, $header->samesite, $header->priority); } if (is_object($result)) { $result = $result instanceof ToArray ? $result->toArray() : get_object_vars($result); } if (is_array($result)) { $result = json_encode($result, JSON_UNESCAPED_UNICODE); } return $response->end($result); } /** * @param string $host */ public function setHost(string $host): void { $this->host = $host; } /** * @param int $port */ public function setPort(int $port): void { $this->port = $port; } /** * @param bool $isSsl */ public function setIsSsl(bool $isSsl): void { $this->isSsl = $isSsl; } /** * @param bool $reuse_port */ public function setReusePort(bool $reuse_port): void { $this->reuse_port = $reuse_port; } }