<?phpnamespace App\Net;use App\Chat\ChatServer;use App\Chat\Frame;/** * The chat on a TCP port: one non-blocking listening socket, stream_select() over every client, frames cut at * line ends and handed to the ChatServer. This runs on a server (Docker, a service), never in a browser. */final class TcpServer{ public const MAX_LINE = 8192; public const MAX_CLIENTS = 2000; /** @var resource|null */ private $listener = null; /** @var resource|null the WebSocket port (php-wasm-network clients, behind the web server on 443) */ private $wsListener = null; /** @var array<int, TcpSession|WsSession> by resource id */ private array $sessions = []; private bool $running = false; /** @var \Closure(string): void */ private \Closure $log; public function __construct(private readonly ChatServer $chat, ?\Closure $log = null) { $this->log = $log ?? static function (string $line): void { }; } public function listen(string $host, int $port): string { $this->listener = self::open($host, $port); return (string)stream_socket_get_name($this->listener, false); } /** The WebSocket port, beside the TCP one. */ public function listenWebSocket(string $host, int $port): string { $this->wsListener = self::open($host, $port); return (string)stream_socket_get_name($this->wsListener, false); } /** @return resource */ private static function open(string $host, int $port) { $address = (str_contains($host, ':') ? "[$host]" : $host) . ':' . $port; $listener = @stream_socket_server("tcp://$address", $errno, $error); if ($listener === false) { throw new \RuntimeException("cannot listen on $address: $error ($errno)"); } stream_set_blocking($listener, false); return $listener; } /** * Serves until stop(). $tick runs about once a second (status file, signals). * * @param \Closure(): void|null $tick */ public function run(?\Closure $tick = null, float $tickEvery = 1.0): void { $this->running = true; $nextTick = microtime(true) + $tickEvery; while ($this->running) { $this->pollOnce(min(1.0, max(0.0, $nextTick - microtime(true)))); if (microtime(true) >= $nextTick) { $nextTick = microtime(true) + $tickEvery; $this->chat->tick(); foreach ($this->sessions as $session) { if ($session instanceof WsSession) { $session->keepAlive(microtime(true)); } } if ($tick !== null) { $tick(); } } } foreach ($this->sessions as $session) { $this->chat->disconnect($session); $session->close(); } $this->sessions = []; foreach ([$this->listener, $this->wsListener] as $listener) { if (is_resource($listener)) { fclose($listener); } } } public function stop(): void { $this->running = false; } /** One pass: accept, read, write. Public for the tests (a server and clients in one process). */ public function pollOnce(float $timeout): void { $read = array_values(array_filter([$this->listener, $this->wsListener], 'is_resource')); $write = []; foreach ($this->sessions as $session) { $read[] = $session->stream; if ($session->hasOutput()) { $write[] = $session->stream; } } $except = null; $seconds = (int)$timeout; $micro = (int)(($timeout - $seconds) * 1_000_000); if (@stream_select($read, $write, $except, $seconds, $micro) === false) { return; // interrupted by a signal } foreach ($read as $stream) { if ($stream === $this->listener || $stream === $this->wsListener) { $this->accept($stream, $stream === $this->wsListener); continue; } $session = $this->sessions[(int)$stream] ?? null; if ($session !== null) { $this->readFrom($session); } } foreach ($write as $stream) { $session = $this->sessions[(int)$stream] ?? null; $session?->flush(); } foreach ($this->sessions as $id => $session) { if ($session->isClosed()) { $this->chat->disconnect($session); unset($this->sessions[$id]); } } } /** @param resource $listener */ private function accept($listener, bool $ws): void { $stream = @stream_socket_accept($listener, 0, $peer); if ($stream === false) { return; } if (count($this->sessions) >= self::MAX_CLIENTS) { fwrite($stream, Frame::encode('12')); // 12: the server is full (the applet's message) fclose($stream); return; } stream_set_blocking($stream, false); if ($ws) { // the chat meets it once it joined (WsSession::control) $this->sessions[(int)$stream] = new WsSession($stream, (string)$peer, $this->chat); return; } $session = new TcpSession($stream, (string)$peer); $this->sessions[(int)$stream] = $session; $this->chat->connect($session); $session->flush(); } private function readFrom(TcpSession|WsSession $session): void { $data = @fread($session->stream, 65536); if ($data === false || ($data === '' && feof($session->stream))) { $session->close(); return; } if ($session instanceof WsSession) { $session->read($data); return; } $session->buffer .= $data; foreach (Frame::decode($session->buffer) as [$code, $fields]) { if ($session->isClosed()) { return; } $this->chat->receive($session, $code, $fields); } if (strlen($session->buffer) > self::MAX_LINE) { ($this->log)("line too long from {$session->address()}"); $this->chat->disconnect($session); $session->close(); } } public function clientCount(): int { return count($this->sessions); }}