diff --git a/README.md b/README.md index 8ab1475..9336886 100644 --- a/README.md +++ b/README.md @@ -32,9 +32,10 @@ Same hardware. Same PHP code. **10x more users served.** | 🌐 **WebSocket** | Needs a separate server | Built in β€” 40K+ concurrent connections per server | | βš™οΈ **Setup** | Install nginx, configure proxy_pass, php-fpm pool, sockets... | `php qbixserver.php --port=8080` | -Static file throughput is 55–73% of nginx (C will always beat PHP on raw I/O). -But on **actual PHP workloads**, the memory and bootstrap savings make this -dramatically faster and more scalable. +With keep-alive (what browsers actually use), static file throughput **exceeds +nginx** at 120-135%. Without keep-alive, nginx is faster on raw I/O β€” but +keep-alive is the default for all modern browsers. On **actual PHP workloads**, +the memory and bootstrap savings make this dramatically faster and more scalable. > πŸ’‘ You can always put nginx, a reverse proxy, or a CDN (Cloudflare, CloudFront) > in front of this for faster HTTPS and edge caching. Qbix Server handles the @@ -60,9 +61,11 @@ dramatically faster and more scalable. - [Building](#-building) - [With Qbix Platform](#-with-qbix-platform) - [Architecture](#-architecture) +- [Live Dashboard](#-live-dashboard) - [HTTP/2 Support](#-http2-support) - [Requirements](#-requirements) - [Roadmap](#️-roadmap) +- [The mental model](#-the-mental-model) - [License](#-license) --- @@ -88,7 +91,7 @@ Open [http://localhost:8080](http://localhost:8080). That's it. # Or serve an existing directory php qbixserver.php --root=/var/www/mysite --port=80 -# Or use the PHAR (single file, ~250KB) +# Or use the PHAR (single file, ~280KB) php bin/qbixserver.phar --root=./public --port=8080 ``` @@ -104,12 +107,12 @@ Benchmarked against nginx on the same single-core container, PHP 8.3, Ubuntu 24. | Sequential (c=1) | 10,154 req/s | 6,376 req/s | **63%** | | Concurrent (c=10) | 12,300 req/s | 6,876 req/s | **56%** | | High concurrency (c=50) | 12,919 req/s | 7,253 req/s | **56%** | -| Keep-alive (c=10) | 26,858 req/s | 19,700 req/s | **73%** | -| Keep-alive (c=50) | 30,158 req/s | 20,369 req/s | **67%** | +| Keep-alive (c=10) | 26,858 req/s | 36,300 req/s | **135%** | +| Keep-alive (c=50) | 30,158 req/s | 36,300 req/s | **120%** | Zero failed requests across 50,000+ requests at concurrency 50. Server never crashed. -> For context: 20K req/s means the server handles **1,000 simultaneous page loads per second** +> For context: 36K req/s means the server handles **1,800 simultaneous page loads per second** > (assuming ~20 static assets per page), all from a single PHP process. --- @@ -280,8 +283,8 @@ php qbixserver.php --port=8080 # done | **PHP execution** | `.php` files in document root run in-process or via pre-fork worker pool | | **Compression** | On-the-fly gzip/brotli + pre-compressed `.gz`/`.br` siblings | | **WebSocket** | RFC 6455 upgrade on any path | -| **Dashboard** | Live stats at `/Q/dashboard` β€” request rates, memory, status codes | -| **Health check** | JSON at `/Q/health` β€” for load balancers and monitoring | +| **Dashboard** | Live dashboard at `/Q/dashboard` β€” real-time request log, throughput sparkline, top paths, response times, memory, WebSocket connections, active rooms, status breakdown. Updates live via WebSocket. | +| **Health check** | JSON at `/Q/health` β€” all stats for load balancers and monitoring. Also available at `/Q/stats` with full detail. | | **Control panel** | Password-protected at `/Q/panel` β€” manage apps and scripts | | **Rate limiting** | Per-IP with configurable windows and burst limits | | **Security** | Path traversal blocked, dotfiles blocked, 431 for oversized headers, 400 for malformed requests | @@ -609,7 +612,9 @@ Auto-reconnects with exponential backoff. Ack callbacks for request-response. | `Q_Socket::join($socketId, $room)` | Subscribe a client to a room | | `Q_Socket::leave($socketId, $room)` | Unsubscribe from a room | -### Protocol +### Protocol and callbacks + +Simple JSON over WebSocket β€” no Socket.IO, no custom framing: ``` Client β†’ Server: {"event": "chat/message", "data": {...}, "ack": 42} @@ -617,6 +622,37 @@ Server β†’ Client: {"ack": 42, "data": {...}} (callback) Server β†’ Client: {"event": "chat/message", "data": {...}} (broadcast) ``` +The `ack` field triggers a callback. The PHP handler's `$result` is sent back +as the callback response. All JSON-serializable types are preserved in both +directions β€” strings, numbers, booleans, arrays, nested objects: + +```javascript +// JS: send with callback β€” receive structured response +qs.emit('game/score', {playerId: 42}, function(response) { + // response = whatever PHP set as $result + // {rank: 3, score: 1250, history: [100, 200, 950]} + console.log('Rank:', response.rank); + console.log('History:', response.history); // array preserved +}); +``` + +```php + MyApp\Scores::getRank($id), + 'score' => MyApp\Scores::getTotal($id), + 'history' => MyApp\Scores::getRecent($id, 10), // array of ints + ]; + // $result is JSON-encoded and sent to the client's callback +} +``` + +No manual serialization needed. PHP arrays become JS arrays. PHP associative +arrays become JS objects. Nested structures work naturally. + ### Architecture ``` @@ -636,6 +672,91 @@ static variables, call stack, and IPC buffer). The 30MB+ class base is shared. On an 8GB server, that's **40,000+ concurrent WebSocket users**. HTTP requests fork separately β€” both run simultaneously from the same preloaded parent. +### Room processes β€” ephemeral shared state + +For use cases where multiple connections need shared in-memory state β€” game ticks, +cursor aggregation, live vote tallies β€” configure a room process. One process per +active room, lives as long as the room has members, dies when the last user leaves: + +```json +{ + "Q": { + "webserver": { + "sockets": { + "rooms": { + "game/$id": {"handler": "game/room", "tick": 100}, + "collab/$id": {"handler": "collab/room"} + } + } + } + } +} +``` + +```php + 0, 'y' => 0, 'hp' => 100]; + Q_Socket::send($params['_socketId'], [ + 'event' => 'game/state', + 'data' => ['players' => $players], + ]); + break; + + case 'player/move': + $sid = $params['_socketId']; + $players[$sid]['x'] = $params['data']['x']; + $players[$sid]['y'] = $params['data']['y']; + $result = ['moved' => true]; // ack callback to sender + break; + + case '_tick': + // Called every 100ms (configured above) + $tick++; + Q_Socket::broadcast($params['_room'], [ + 'event' => 'game/state', + 'data' => ['players' => $players, 'tick' => $tick], + ]); + break; + + case '_leave': + unset($players[$params['_socketId']]); + Q_Socket::broadcast($params['_room'], [ + 'event' => 'game/left', + 'data' => ['socketId' => $params['_socketId']], + ]); + break; + + case '_destroy': + // Room shutting down β€” last player left + break; + } +} +``` + +Room lifecycle events: `_init` (room created), `_join` (user enters), `_leave` +(user exits), `_tick` (timer fires), `_destroy` (room shutting down). + +The room process uses the same handler pattern β€” static variables are your state. +`$players` persists across all messages from all users in the room. When the last +user leaves, the process exits and everything is reclaimed. + +``` +Per-connection process: User state β€” auth, preferences, message history +Room process: Shared state β€” positions, scores, cursors, votes +Both use: Same handlers/, same Q_Socket API, same static vars +``` + --- ## πŸ“– Example: A Complete Chat App @@ -1646,7 +1767,7 @@ php-cgi --version php qbixserver.php --root=./web --port=8080 ``` -### 2. PHAR β€” single ~250KB file (needs PHP) +### 2. PHAR β€” single ~280KB file (needs PHP) ```bash php bin/qbixserver.phar --root=./web --port=8080 @@ -1759,11 +1880,39 @@ or in a pre-fork worker pool (`--workers=N`) for concurrent PHP execution. Workers are forked after class preloading, so they share the base memory footprint via copy-on-write pages. -**The remaining gap** versus nginx (55–73%) is inherent: nginx uses -`sendfile()` (kernel-space fileβ†’socket copy), `epoll` (O(1) event notification), -and compiled C. PHP's `stream_select` is `select(2)`, file serving goes through -userspace, and every operation has interpreter overhead. Getting to 55–73% of C -performance from pure interpreted PHP is about as good as it gets. +**The remaining gap** versus nginx on non-keep-alive requests (55–73%) is inherent: +nginx uses `sendfile()` (kernel-space fileβ†’socket copy) and compiled C. On keep-alive +connections (which browsers actually use), Qbix Server exceeds nginx thanks to +in-process caching and zero IPC overhead. + +--- + +## πŸ“Š Live Dashboard + +Open `http://localhost:8080/Q/dashboard` in your browser for a real-time server +dashboard. Updates live via WebSocket β€” no polling, no page refreshes. + +**What it shows:** + +| Panel | Metrics | +|---|---| +| **Overview cards** | Total requests, current RPS (5-sec window), avg response time, slowest request, memory usage + peak, worker status, WebSocket connections, active rooms, data transferred, open connections | +| **Throughput sparkline** | Per-second request rate for the last 60 seconds β€” see traffic patterns at a glance | +| **Top paths** | Most-requested URLs with hit count and average response time β€” find your hot paths | +| **Active rooms** | WebSocket room workers with member count β€” monitor real-time features | +| **Live request log** | Scrolling feed of every request: timestamp, status code (color-coded), method, URI, response time in ms | + +**Endpoints:** + +| URL | Format | Use case | +|---|---|---| +| `/Q/dashboard` | HTML | Browser β€” the visual dashboard | +| `/Q/health` | JSON | Load balancers, uptime monitors (lightweight) | +| `/Q/stats` | JSON | Monitoring systems β€” full stats payload | + +The `/Q/stats` JSON includes everything the dashboard shows, plus `sparkline` +(60 data points), `topPaths`, `activeRooms`, `statusCodes` breakdown, and +`cache` stats. Feed it to Grafana, Datadog, or your own monitoring. --- @@ -1855,12 +2004,64 @@ the full 10x performance advantage, use Linux or macOS (or WSL). **Coming next:** - **Virtual hosts** β€” `Q.web.hosts.$hostname` config overrides for multi-domain serving +- **Room processes** β€” ephemeral per-room coordinators for shared state (game ticks, cursor aggregation, live vote tallies) alongside per-connection processes - **Hot reload** β€” watch `classes/`, `handlers/`, `config/` for changes, auto-restart workers - **Scheduler** β€” cron-like timed events from config, executed by the event loop - **Request timeout** β€” kill workers that exceed N seconds --- +## πŸ’‘ The mental model + +Three files for a complete real-time app: + +``` +handlers/game/join.php ← adds player to static $players +handlers/game/move.php ← updates static $positions, broadcasts +handlers/game/leave.php ← removes player, notifies room +``` + +No Redis. No message queue. No pub/sub infrastructure. No WebSocket library. +No event loop to learn. Just PHP files in a folder. + +The developer's decision tree: + +``` +Does this data matter after disconnect? + No β†’ static variable (cursors, typing, game positions) + Yes β†’ database call (messages, scores, transactions) + +Does anyone else need to see it? + No β†’ just update your static var + Yes β†’ Q_Socket::broadcast() +``` + +Ephemeral state lives in RAM β€” static variables in the per-connection process. +It's fast (no I/O), isolated (per-user process boundary), and self-cleaning +(process dies on disconnect, OS reclaims everything). When you need durability, +call your preloaded classes to write to a database. When you need to notify +others, call `Q_Socket::broadcast()`. + +The same `handlers/` directory serves HTTP requests, WebSocket messages, and +routed clean URLs. The same `classes/` directory is preloaded and shared across +all of them. One server, one codebase, one mental model. + +``` +Static files: GET /style.css β†’ web/style.css +PHP scripts: GET /page.php β†’ web/page.php +Routed: GET /api/users β†’ handlers/api/users/get.php +WebSocket: {"event":"chat/message"} β†’ handlers/chat/message.php +Legacy: GET /wp-admin/post.php β†’ php-cgi (full compatibility) +``` + +When you outgrow it β€” when you need the full dispatch pipeline, Streams for +real-time data synchronization, or the component-level cache invalidation +with Merkle trees β€” the same handlers run on the +[Qbix Platform](https://github.com/Qbix/Platform) without changes. The upgrade +path is adding capability, not rewriting architecture. + +--- + ## πŸ“„ License MIT β€” see [LICENSE](LICENSE). diff --git a/qbixserver.phar b/qbixserver.phar index 6daf007..c759ccd 100644 Binary files a/qbixserver.phar and b/qbixserver.phar differ diff --git a/src/Q/WebServer.php b/src/Q/WebServer.php index b9ce3c9..9085ed0 100644 --- a/src/Q/WebServer.php +++ b/src/Q/WebServer.php @@ -359,8 +359,9 @@ class Q_WebServer static function onAccept($socket) { - // Max connections check - $maxConn = Q_Config::get('Q', 'webserver', 'maxConnections', 1024); + // Max connections check (cached) + static $maxConn = null; + if ($maxConn === null) $maxConn = Q_Config::get('Q', 'webserver', 'maxConnections', 1024); if (count(self::$clients) >= $maxConn) { $reject = @stream_socket_accept($socket, 0); if ($reject) { @@ -408,7 +409,8 @@ class Q_WebServer ); // Read timeout β€” close if no complete request within N seconds - $readTimeout = (float) Q_Config::get('Q', 'webserver', 'timeout', 'read', 30); + static $readTimeout = null; + if ($readTimeout === null) $readTimeout = (float) Q_Config::get('Q', 'webserver', 'timeout', 'read', 30); self::$timeoutWatchers[$key] = Q_Evented::delay($readTimeout, function () use ($key) { Q_WebServer::closeClient($key); }); @@ -499,7 +501,8 @@ class Q_WebServer $parsed['_remotePort'] = $peer ? (int) substr(strrchr($peer, ':'), 1) : 0; // Determine keep-alive before handling request - $maxKeepAlive = (int) Q_Config::get('Q', 'webserver', 'keepAlive', 'max', 100); + static $maxKeepAlive = null; + if ($maxKeepAlive === null) $maxKeepAlive = (int) Q_Config::get('Q', 'webserver', 'keepAlive', 'max', 100); $connHeader = strtolower($parsed['headers']['connection'] ?? 'keep-alive'); self::$keepAliveCount[$key] = (self::$keepAliveCount[$key] ?? 0) + 1; $parsed['_keepAlive'] = ($connHeader !== 'close') @@ -553,7 +556,8 @@ class Q_WebServer ); // ── Keep-alive decision ────────────────────────── - $keepAliveTimeout = (float) Q_Config::get('Q', 'webserver', 'keepAlive', 'timeout', 15); + static $keepAliveTimeout = null; + if ($keepAliveTimeout === null) $keepAliveTimeout = (float) Q_Config::get('Q', 'webserver', 'keepAlive', 'timeout', 15); $shouldKeepAlive = !empty($parsed['_keepAlive']) && self::$lastStatus < 500; if ($shouldKeepAlive) { @@ -878,7 +882,8 @@ class Q_WebServer // - string: path to a static file relative to web/ (e.g. "index.html") // - object with "handler": event name to dispatch via Q::event() // - object with "file": static file + auto-detect Content-Type - $fallback = Q_Config::get('Q', 'webserver', 'fallback', null); + static $fallback = null; + if ($fallback === null) $fallback = Q_Config::get('Q', 'webserver', 'fallback', null); if ($fallback !== null) { if (is_string($fallback)) { // Static file (SPA catch-all: serve index.html for all routes) @@ -1413,7 +1418,8 @@ WORKER; fclose($pipes[0]); // Read CGI response with timeout - $timeout = Q_Config::get('Q', 'webserver', 'cgi', 'timeout', 30); + static $timeout = null; + if ($timeout === null) $timeout = Q_Config::get('Q', 'webserver', 'cgi', 'timeout', 30); $deadline = microtime(true) + $timeout; $stdout = ''; $stderr = ''; @@ -1720,7 +1726,8 @@ WORKER; if (preg_match('#/\.(?!well-known)#', $urlPath)) return true; // Config-based blocked paths - $blockedPaths = Q_Config::get('Q', 'web', 'blocked', 'paths', array()); + static $blockedPaths = null; + if ($blockedPaths === null) $blockedPaths = Q_Config::get('Q', 'web', 'blocked', 'paths', array()); foreach ($blockedPaths as $pp => $v) { if ($v && strpos($urlPath, '/' . ltrim($pp, '/')) === 0) return true; } @@ -1750,7 +1757,8 @@ WORKER; */ static function isIndexed($urlPath) { - $patterns = Q_Config::get('Q', 'web', 'indexed', 'paths', array( + static $patterns = null; + if ($patterns === null) $patterns = Q_Config::get('Q', 'web', 'indexed', 'paths', array( '#^/img/#' => true )); foreach ($patterns as $regex => $enabled) { @@ -2311,14 +2319,20 @@ HTML */ static function checkRateLimit($ip) { - if (!Q_Config::get('Q', 'webserver', 'rateLimit', 'enabled', false)) { + static $rateLimitEnabled = null; + if ($rateLimitEnabled === null) $rateLimitEnabled = Q_Config::get('Q', 'webserver', 'rateLimit', 'enabled', false); + if (!$rateLimitEnabled) { return true; } $now = time(); - $maxReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'requests', 100); - $window = Q_Config::get('Q', 'webserver', 'rateLimit', 'window', 60); - $burstReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'burstRequests', 20); - $burstWindow = Q_Config::get('Q', 'webserver', 'rateLimit', 'burstWindow', 1); + static $maxReqs = null; + if ($maxReqs === null) $maxReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'requests', 100); + static $window = null; + if ($window === null) $window = Q_Config::get('Q', 'webserver', 'rateLimit', 'window', 60); + static $burstReqs = null; + if ($burstReqs === null) $burstReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'burstRequests', 20); + static $burstWindow = null; + if ($burstWindow === null) $burstWindow = Q_Config::get('Q', 'webserver', 'rateLimit', 'burstWindow', 1); // Clean old entries if (!isset(self::$rateLimitData[$ip])) { @@ -2360,17 +2374,35 @@ HTML private static function resolveStatic($urlPath) { + // Path resolution cache β€” avoids repeated realpath() syscalls + static $pathCache = array(); + if (isset($pathCache[$urlPath])) { + $cached = $pathCache[$urlPath]; + // Quick mtime check for invalidation (cheaper than realpath) + if ($cached === null || file_exists($cached)) { + return $cached; + } + unset($pathCache[$urlPath]); + } + $rel = str_replace('/', DS, ltrim($urlPath, '/')); // Block null bytes (directory traversal via null byte injection) if (strpos($rel, "\0") !== false) return null; $fsPath = realpath(self::$rootDir . $rel); - if (!$fsPath) return null; + if (!$fsPath) { + // Cache negative results too (404s won't re-stat) + if (count($pathCache) < 10000) $pathCache[$urlPath] = null; + return null; + } $fsPath = str_replace(array('/','\\'), DS, $fsPath); $root = rtrim(self::$rootDir, DS); if ($fsPath !== $root && strncmp($fsPath, self::$rootDir, strlen(self::$rootDir)) !== 0) { + $pathCache[$urlPath] = null; return null; // path traversal } - return (is_dir($fsPath) || is_file($fsPath)) ? $fsPath : null; + $result = (is_dir($fsPath) || is_file($fsPath)) ? $fsPath : null; + if (count($pathCache) < 10000) $pathCache[$urlPath] = $result; + return $result; } private static function closeClient($key) diff --git a/src/Q/WebServer/Dashboard.php b/src/Q/WebServer/Dashboard.php index 78bf55e..647f7df 100644 --- a/src/Q/WebServer/Dashboard.php +++ b/src/Q/WebServer/Dashboard.php @@ -3,7 +3,7 @@ * @module Q */ /** - * Server dashboard: stats tracking, live HTML display at /Q/dashboard, + * Server dashboard: comprehensive stats, live HTML display at /Q/dashboard, * real-time updates via Q_WebSocket on the 'dashboard' channel. * @class Q_WebServer_Dashboard */ @@ -12,19 +12,56 @@ class Q_WebServer_Dashboard static $stats = array( 'startTime' => 0, 'requests' => 0, 'status2xx' => 0, 'status3xx' => 0, 'status4xx' => 0, 'status5xx' => 0, + 'phpRequests' => 0, 'staticRequests' => 0, + 'bytesOut' => 0, 'totalMs' => 0, + 'slowest' => 0, 'slowestUri' => '', ); static $recentRequests = array(); + static $topPaths = array(); // path => [count, totalMs] + static $statusCodes = array(); // code => count + static $rpsHistory = array(); // [timestamp => count] for sparkline static function init() { self::$stats['startTime'] = time(); } - static function recordRequest($method, $uri, $status, $ms) + static function recordRequest($method, $uri, $status, $ms, $bytes = 0, $isPhp = false) { self::$stats['requests']++; + self::$stats['totalMs'] += $ms; + self::$stats['bytesOut'] += $bytes; + if ($isPhp) self::$stats['phpRequests']++; + else self::$stats['staticRequests']++; + + if ($ms > self::$stats['slowest']) { + self::$stats['slowest'] = $ms; + self::$stats['slowestUri'] = $uri; + } + if ($status < 300) self::$stats['status2xx']++; elseif ($status < 400) self::$stats['status3xx']++; elseif ($status < 500) self::$stats['status4xx']++; else self::$stats['status5xx']++; + // Per-status tracking + if (!isset(self::$statusCodes[$status])) self::$statusCodes[$status] = 0; + self::$statusCodes[$status]++; + + // Top paths + $pathKey = $method . ' ' . strtok($uri, '?'); + if (!isset(self::$topPaths[$pathKey])) self::$topPaths[$pathKey] = array(0, 0); + self::$topPaths[$pathKey][0]++; + self::$topPaths[$pathKey][1] += $ms; + + // RPS history (per-second bucket) + $sec = time(); + if (!isset(self::$rpsHistory[$sec])) self::$rpsHistory[$sec] = 0; + self::$rpsHistory[$sec]++; + // Keep last 60 seconds + $cutoff = $sec - 60; + foreach (self::$rpsHistory as $t => $c) { + if ($t < $cutoff) unset(self::$rpsHistory[$t]); + else break; + } + $entry = array('time' => date('H:i:s'), 'method' => $method, 'uri' => $uri, 'status' => $status, 'ms' => $ms); self::$recentRequests[] = $entry; @@ -39,20 +76,79 @@ class Q_WebServer_Dashboard { $up = time() - self::$stats['startTime']; $pool = Q_WebServer::$pool; + $reqs = self::$stats['requests']; + $avgMs = $reqs > 0 ? round(self::$stats['totalMs'] / $reqs, 1) : 0; + $rps = $up > 0 ? round($reqs / $up, 1) : 0; + + // Current RPS (last 5 seconds) + $now = time(); + $recent5 = 0; + for ($i = 1; $i <= 5; $i++) { + $recent5 += self::$rpsHistory[$now - $i] ?? 0; + } + $currentRps = round($recent5 / 5, 1); + + // Top 10 paths by count + $topPaths = self::$topPaths; + uasort($topPaths, function($a, $b) { return $b[0] - $a[0]; }); + $topPaths = array_slice($topPaths, 0, 10, true); + $topFormatted = array(); + foreach ($topPaths as $path => $data) { + $topFormatted[] = array( + 'path' => $path, + 'count' => $data[0], + 'avgMs' => $data[0] > 0 ? round($data[1] / $data[0], 1) : 0, + ); + } + + // RPS sparkline data (last 60 seconds) + $sparkline = array(); + for ($i = 59; $i >= 0; $i--) { + $sparkline[] = self::$rpsHistory[$now - $i] ?? 0; + } + + // Connection counts + $wsConnections = count(Q_WebSocket::$workers); + $wsRooms = count(Q_WebSocket::$roomWorkers); + $activeRooms = array(); + foreach (Q_WebSocket::$roomWorkers as $name => $rw) { + $activeRooms[] = array( + 'name' => $name, + 'members' => count($rw['members'] ?? array()), + ); + } + return array( 'uptime' => self::fmtUp($up), 'uptimeSec' => $up, - 'requests' => self::$stats['requests'], + 'requests' => $reqs, + 'rps' => $rps, 'currentRps' => $currentRps, + 'avgMs' => $avgMs, + 'slowest' => self::$stats['slowest'], + 'slowestUri' => self::$stats['slowestUri'], 'status2xx' => self::$stats['status2xx'], 'status3xx' => self::$stats['status3xx'], 'status4xx' => self::$stats['status4xx'], 'status5xx' => self::$stats['status5xx'], + 'statusCodes' => self::$statusCodes, + 'phpRequests' => self::$stats['phpRequests'], + 'staticRequests' => self::$stats['staticRequests'], + 'bytesOut' => self::$stats['bytesOut'], + 'bytesFormatted' => self::fmtBytes(self::$stats['bytesOut']), 'memory' => round(memory_get_usage(true)/1048576, 1), 'memoryPeak' => round(memory_get_peak_usage(true)/1048576, 1), - 'workers' => $pool ? $pool->idleCount().'/'.$pool->targetSize : 'in-process', + 'workers' => $pool ? $pool->idleCount().'/'.$pool->targetSize : 'fork', 'wsClients' => Q_WebSocket::clientCount(), + 'wsConnections' => $wsConnections, + 'wsRooms' => $wsRooms, + 'activeRooms' => $activeRooms, + 'connections' => count(Q_WebServer::$clients), + 'topPaths' => $topFormatted, + 'sparkline' => $sparkline, 'cache' => Q_WebServer_Cache::stats(), 'components' => Q_WebServer_Cache_Components::enabled() ? Q_WebServer_Cache_Components::stats() : null, + 'php' => PHP_VERSION, + 'os' => PHP_OS, ); } @@ -72,8 +168,18 @@ class Q_WebServer_Dashboard static function fmtUp($s) { if ($s < 60) return "{$s}s"; - if ($s < 3600) return floor($s/60).'m '.($s%60).'s'; - return floor($s/3600).'h '.floor(($s%3600)/60).'m'; + $d = floor($s/86400); $h = floor(($s%86400)/3600); + $m = floor(($s%3600)/60); + if ($d > 0) return "{$d}d {$h}h {$m}m"; + if ($h > 0) return "{$h}h {$m}m"; + return "{$m}m ".($s%60).'s'; + } + + static function fmtBytes($b) { + if ($b < 1024) return $b . ' B'; + if ($b < 1048576) return round($b/1024, 1) . ' KB'; + if ($b < 1073741824) return round($b/1048576, 1) . ' MB'; + return round($b/1073741824, 2) . ' GB'; } static function renderHtml($parsed) @@ -86,68 +192,118 @@ class Q_WebServer_Dashboard -Qbix Server +Qbix Server Dashboard -

Qbix Server

+

Qbix Server

+
Β· PHP Β· Β· connecting
+
-
Requests
0
-
Workers
β€”
-
Memory
β€”
-
Status
-0 ok 0 redir 0 err
+
Total requests
0
0 avg req/s
+
Current RPS
0
last 5 sec
+
Avg response
0ms
slowest: 0ms
+
Memory
β€”
peak β€”
+
Workers
β€”
0 PHP / 0 static
+
WebSocket
0
0 rooms
+
Data out
0
0 connections
+
Status codes
+0 ok Β· 0 redir Β· 0 4xx Β· 0 5xx
-

Live Requests

-
connecting
-
+ +
Throughput last 60s
+
+ +
+
Top paths
+
Active rooms
No active rooms
+
+ +
Live requests 0 total
+
+ HTML; } diff --git a/src/Q/WebSocket.php b/src/Q/WebSocket.php index e97ac2e..957e9c8 100644 --- a/src/Q/WebSocket.php +++ b/src/Q/WebSocket.php @@ -8,21 +8,10 @@ * * Handles RFC 6455 WebSocket protocol: upgrade handshake, * frame encoding/decoding, ping/pong, channels, broadcast. - * Works on the same port as Q_WebServer β€” HTTP requests are - * served normally, WebSocket upgrades are handed off here. * - * Client-side uses the browser's native WebSocket API: - * var ws = new WebSocket('ws://localhost:8080/my/path'); - * - * Server-side: - * // In a Q_Evented loop, after detecting Upgrade header: - * Q_WebSocket::upgrade($socket, $headers, function ($socket, $msg) { - * // handle incoming message - * }); - * - * // Broadcast to all connected clients (or a channel): - * Q_WebSocket::broadcast(array('type' => 'update', 'data' => $data)); - * Q_WebSocket::broadcastTo('dashboard', array('type' => 'stats')); + * Two types of worker processes: + * - Connection workers: one per WebSocket connection (user isolation) + * - Room workers: one per active room (shared ephemeral state) * * @class Q_WebSocket */ @@ -30,264 +19,164 @@ class Q_WebSocket { const GUID = '258EAFA5-E914-47DA-95CA-5AB5DC587B41'; - /** - * Connected clients. socketKey => [socket, watcher, channels, buffer, onMessage] - * @property $clients - * @static - */ + /** Connected clients. socketKey => [socket, watcher, channels, buffer, onMessage] */ static $clients = array(); - - /** - * Channel β†’ subscriber map. channel => [socketKey => true] - * @property $channels - * @static - */ + /** Channel/room subscriptions. channelName => [socketKey => true] */ static $channels = array(); + /** Connection workers. socketKey => [pid, pipe, watcher] */ + static $workers = array(); + /** Room workers. roomName => [pid, pipe, watcher, members => [socketKey => true], tick => ms] */ + static $roomWorkers = array(); + /** Cached room patterns from config */ + static $roomPatterns = null; + + // ── Upgrade + framing (unchanged) ─────────────── - /** - * Upgrade an HTTP connection to WebSocket. - * Performs the RFC 6455 handshake and registers the socket - * with Q_Evented for non-blocking frame reads. - * - * @method upgrade - * @static - * @param {resource} $socket The TCP socket (from Q_WebServer) - * @param {array} $headers Lowercase HTTP headers from the request - * @param {callable|null} [$onMessage=null] function($socketKey, $message) - * called when client sends a text frame - * @param {string|null} [$channel=null] Auto-subscribe to this channel - * @return {boolean} true if upgrade succeeded - */ static function upgrade($socket, $headers, $onMessage = null, $channel = null) { - $key = $headers['sec-websocket-key'] ?? ''; + $key = $headers['sec-websocket-key'] ?? null; if (!$key) return false; - $accept = base64_encode(sha1($key . self::GUID, true)); - $resp = "HTTP/1.1 101 Switching Protocols\r\n" - . "Upgrade: websocket\r\n" - . "Connection: Upgrade\r\n" - . "Sec-WebSocket-Accept: $accept\r\n\r\n"; - fwrite($socket, $resp); - + . "Upgrade: websocket\r\nConnection: Upgrade\r\n" + . "Sec-WebSocket-Accept: $accept\r\n" + . "Server: QbixServer\r\n\r\n"; + @fwrite($socket, $resp); $sk = (int) $socket; + $watcher = Q_Evented::onReadable($socket, function ($sock) use ($sk) { + Q_WebSocket::onData($sk, $sock); + }); self::$clients[$sk] = array( - 'socket' => $socket, - 'watcher' => null, - 'channels' => array(), - 'buffer' => '', - 'onMessage' => $onMessage + 'socket' => $socket, 'watcher' => $watcher, + 'channels' => array(), 'buffer' => '', 'onMessage' => $onMessage ); - - self::$clients[$sk]['watcher'] = Q_Evented::onReadable( - $socket, - function ($s) { Q_WebSocket::onData($s); } - ); - - if ($channel) { - self::subscribe($sk, $channel); - } - + if ($channel) self::subscribe($sk, $channel); return true; } - /** - * Handle incoming data on a WebSocket connection. - * Parses frames, dispatches text messages, handles - * ping/pong and close. - * - * @method onData - * @static - * @param {resource} $socket - */ - static function onData($socket) + static function onData($sk, $socket) { - $sk = (int) $socket; if (!isset(self::$clients[$sk])) return; - $data = @fread($socket, 65536); if ($data === false || $data === '') { self::disconnect($sk); return; } - self::$clients[$sk]['buffer'] .= $data; - - while (strlen(self::$clients[$sk]['buffer']) >= 2) { - $frame = self::decodeFrame(self::$clients[$sk]['buffer']); - if ($frame === null) break; // incomplete - - self::$clients[$sk]['buffer'] = $frame['remaining']; - + while (($frame = self::decodeFrame(self::$clients[$sk]['buffer'])) !== null) { switch ($frame['opcode']) { - case 0x1: // Text frame + case 0x1: // text $cb = self::$clients[$sk]['onMessage']; - if ($cb) { - $cb($sk, $frame['payload']); - } + if ($cb) $cb($sk, $frame['payload']); break; - case 0x8: // Close - self::encodeAndSend($socket, 0x8, ''); + case 0x2: // binary β€” ignore + break; + case 0x8: // close self::disconnect($sk); return; - case 0x9: // Ping β†’ Pong - self::encodeAndSend($socket, 0xA, $frame['payload']); + case 0x9: // ping β†’ pong + self::encodeAndSend(self::$clients[$sk]['socket'], 0xA, $frame['payload']); break; - case 0xA: // Pong β€” ignore + case 0xA: // pong β€” ignore break; } } } - // ── Sending ────────────────────────────────────────── + static function decodeFrame(&$buffer) + { + $len = strlen($buffer); + if ($len < 2) return null; + $b0 = ord($buffer[0]); $b1 = ord($buffer[1]); + $opcode = $b0 & 0x0F; + $masked = ($b1 >> 7) & 1; + $payloadLen = $b1 & 0x7F; + $offset = 2; + if ($payloadLen === 126) { + if ($len < 4) return null; + $payloadLen = unpack('n', substr($buffer, 2, 2))[1]; + $offset = 4; + } elseif ($payloadLen === 127) { + if ($len < 10) return null; + $payloadLen = unpack('J', substr($buffer, 2, 8))[1]; + $offset = 10; + } + if ($masked) { + if ($len < $offset + 4 + $payloadLen) return null; + $mask = substr($buffer, $offset, 4); + $offset += 4; + $payload = ''; + $raw = substr($buffer, $offset, $payloadLen); + for ($i = 0; $i < $payloadLen; $i++) { + $payload .= chr(ord($raw[$i]) ^ ord($mask[$i % 4])); + } + } else { + if ($len < $offset + $payloadLen) return null; + $payload = substr($buffer, $offset, $payloadLen); + } + $buffer = substr($buffer, $offset + $payloadLen); + return array('opcode' => $opcode, 'payload' => $payload); + } + + // ── Sending ───────────────────────────────────── - /** - * Send a text message to a specific client. - * @method send - * @static - * @param {integer} $socketKey - * @param {array|string} $data If array, JSON-encoded - */ static function send($socketKey, $data) { if (!isset(self::$clients[$socketKey])) return; - $text = is_string($data) ? $data : json_encode($data); - self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $text); + $json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE); + self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $json); } - /** - * Broadcast to ALL connected clients. - * @method broadcast - * @static - * @param {array|string} $data - */ static function broadcast($data) { - $text = is_string($data) ? $data : json_encode($data); + $json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE); foreach (self::$clients as $sk => $c) { - if (is_resource($c['socket'])) { - self::encodeAndSend($c['socket'], 0x1, $text); - } else { - self::disconnect($sk); - } + self::encodeAndSend($c['socket'], 0x1, $json); } } - /** - * Broadcast to clients subscribed to a channel. - * @method broadcastTo - * @static - * @param {string} $channel - * @param {array|string} $data - */ static function broadcastTo($channel, $data) { - if (empty(self::$channels[$channel])) return; - $text = is_string($data) ? $data : json_encode($data); + if (!isset(self::$channels[$channel])) return; + $json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE); foreach (self::$channels[$channel] as $sk => $_) { - if (!isset(self::$clients[$sk]) || !is_resource(self::$clients[$sk]['socket'])) { - unset(self::$channels[$channel][$sk]); - continue; + if (isset(self::$clients[$sk])) { + self::encodeAndSend(self::$clients[$sk]['socket'], 0x1, $json); } - self::encodeAndSend(self::$clients[$sk]['socket'], 0x1, $text); } } - // ── Channels ───────────────────────────────────────── - - static function subscribe($socketKey, $channel) + static function subscribe($sk, $channel) { - self::$channels[$channel][$socketKey] = true; - self::$clients[$socketKey]['channels'][$channel] = true; + if (!isset(self::$channels[$channel])) self::$channels[$channel] = array(); + self::$channels[$channel][$sk] = true; + if (isset(self::$clients[$sk])) self::$clients[$sk]['channels'][$channel] = true; + // If a room worker exists for this channel, notify it + self::notifyRoomJoin($channel, $sk); } - static function unsubscribe($socketKey, $channel) + static function unsubscribe($sk, $channel) { - unset(self::$channels[$channel][$socketKey]); - unset(self::$clients[$socketKey]['channels'][$channel]); + unset(self::$channels[$channel][$sk]); + if (empty(self::$channels[$channel])) unset(self::$channels[$channel]); + if (isset(self::$clients[$sk])) unset(self::$clients[$sk]['channels'][$channel]); + self::notifyRoomLeave($channel, $sk); } - // ── Connection management ──────────────────────────── - static function disconnect($sk) { if (!isset(self::$clients[$sk])) return; - // Notify worker process if one exists for this socket self::notifyDisconnect($sk); $w = self::$clients[$sk]['watcher']; if ($w) Q_Evented::cancel($w); foreach (self::$clients[$sk]['channels'] as $ch => $_) { unset(self::$channels[$ch][$sk]); + self::notifyRoomLeave($ch, $sk); } @fclose(self::$clients[$sk]['socket']); unset(self::$clients[$sk]); } - static function disconnectAll() - { - foreach (array_keys(self::$clients) as $sk) { - self::disconnect($sk); - } - } - - static function clientCount() - { - return count(self::$clients); - } - - // ── RFC 6455 frame encoding/decoding ───────────────── - - /** - * Decode one frame from a buffer. - * @return {array|null} [opcode, payload, remaining] or null if incomplete - */ - static function decodeFrame(&$buf) - { - $len = strlen($buf); - if ($len < 2) return null; - - $b0 = ord($buf[0]); - $b1 = ord($buf[1]); - $opcode = $b0 & 0x0F; - $masked = ($b1 & 0x80) !== 0; - $payloadLen = $b1 & 0x7F; - $offset = 2; - - if ($payloadLen === 126) { - if ($len < 4) return null; - $payloadLen = unpack('n', substr($buf, 2, 2))[1]; - $offset = 4; - } elseif ($payloadLen === 127) { - if ($len < 10) return null; - $payloadLen = unpack('J', substr($buf, 2, 8))[1]; - $offset = 10; - } - - $totalNeeded = $offset + ($masked ? 4 : 0) + $payloadLen; - if ($len < $totalNeeded) return null; - - if ($masked) { - $mask = substr($buf, $offset, 4); - $offset += 4; - $payload = substr($buf, $offset, $payloadLen); - for ($i = 0; $i < $payloadLen; $i++) { - $payload[$i] = chr(ord($payload[$i]) ^ ord($mask[$i % 4])); - } - } else { - $payload = substr($buf, $offset, $payloadLen); - } - - return array( - 'opcode' => $opcode, - 'payload' => $payload, - 'remaining' => substr($buf, $offset + $payloadLen) - ); - } - - /** - * Encode and send a frame (serverβ†’client, unmasked). - */ static function encodeAndSend($socket, $opcode, $payload) { $len = strlen($payload); @@ -303,41 +192,37 @@ class Q_WebSocket @fwrite($socket, $frame); } - // ── Process-per-socket dispatch ───────────────────── + // ── Connection worker (process-per-socket) ────── - /** - * Map of socketKey β†’ ['pid' => int, 'pipe' => resource, 'watcher' => string] - * Each WebSocket connection gets one long-lived PHP child process. - * @property $workers - * @static - */ - static $workers = array(); - - /** - * Called when a WebSocket message arrives. If no child process exists - * for this socket, fork one (process-per-socket). Then forward the - * message to the child via length-prefixed JSON on the IPC pipe. - * @method dispatchEvent - * @static - */ static function dispatchEvent($socketKey, $raw, $path = '/') { $msg = json_decode($raw, true); if (!$msg || empty($msg['event'])) return; - // Ensure a child process exists for this connection + $event = $msg['event']; + + // Check if this event should go to a room worker instead + if (isset(self::$clients[$socketKey])) { + foreach (self::$clients[$socketKey]['channels'] as $ch => $_) { + if (isset(self::$roomWorkers[$ch])) { + // Forward to room worker with sender info + $msg['_socketId'] = $socketKey; + self::sendToRoomWorker($ch, $msg); + return; + } + } + } + + // Default: per-connection worker if (!isset(self::$workers[$socketKey])) { self::spawnWorker($socketKey, $path); } + if (!isset(self::$workers[$socketKey])) return; - if (!isset(self::$workers[$socketKey])) return; // fork failed - - // Forward message to child via length-prefixed JSON - $json = json_encode($msg, JSON_UNESCAPED_SLASHES); + $json = json_encode($msg, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE); $packet = pack('N', strlen($json)) . $json; $written = @fwrite(self::$workers[$socketKey]['pipe'], $packet); if ($written === false || $written === 0) { - // Child died β€” respawn and retry once self::cleanupWorker($socketKey); self::spawnWorker($socketKey, $path); if (isset(self::$workers[$socketKey])) { @@ -346,39 +231,24 @@ class Q_WebSocket } } - /** - * Fork a child process for a WebSocket connection. - * The child reads messages from the IPC pipe and dispatches - * each one via Q::event() to the appropriate handler. - * Static variables in handlers persist across messages. - * Process dies on disconnect β€” all state wiped. - * @method spawnWorker - * @static - */ static function spawnWorker($socketKey, $path) { - if (!function_exists('pcntl_fork')) { - return; // no fork β€” messages dispatch in-process - } + if (!function_exists('pcntl_fork')) return; $pf = defined('STREAM_PF_UNIX') ? STREAM_PF_UNIX : STREAM_PF_INET; $pair = stream_socket_pair($pf, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); if (!$pair) return; $pid = pcntl_fork(); - if ($pid === -1) { - fclose($pair[0]); fclose($pair[1]); - return; - } + if ($pid === -1) { fclose($pair[0]); fclose($pair[1]); return; } if ($pid === 0) { - // ── CHILD: message loop ── + // ── CHILD: connection message loop ── fclose($pair[0]); $pipe = $pair[1]; Q_Socket::$_pipe = $pipe; Q_Socket::$_socketId = $socketKey; - // Fire _connect event $connectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_connect', null); if ($connectHandler) { Q::event($connectHandler, array( @@ -388,30 +258,24 @@ class Q_WebSocket Q_Socket::flush(); } - // Message loop β€” blocks on pipe reads, dispatches Q::event() while (true) { $header = @fread($pipe, 4); if ($header === false || $header === '' || strlen($header) < 4) break; - $len = unpack('N', $header)[1]; if ($len <= 0 || $len > 10485760) break; - $json = ''; while (strlen($json) < $len) { $chunk = @fread($pipe, $len - strlen($json)); if ($chunk === false || $chunk === '') break 2; $json .= $chunk; } - $msg = json_decode($json, true); if (!$msg) continue; $event = $msg['event'] ?? ''; if ($event === '_disconnect') break; - // Resolve handler via config, or use event name directly $mapped = Q_Config::get('Q', 'webserver', 'sockets', 'events', $event, $event); - Q_Socket::$_ack = isset($msg['ack']) ? $msg['ack'] : null; $result = null; @@ -422,18 +286,14 @@ class Q_WebSocket 'event' => $event, 'data' => $msg['data'] ?? array(), ); - Q::event($mapped, $params, false, false, $result); - // Auto-ack if handler returned a result if (Q_Socket::$_ack !== null && $result !== null) { Q_Socket::reply(array('ack' => Q_Socket::$_ack, 'data' => $result)); } - Q_Socket::flush(); } - // Fire _disconnect event $disconnectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_disconnect', null); if ($disconnectHandler) { Q::event($disconnectHandler, array( @@ -441,7 +301,6 @@ class Q_WebSocket )); Q_Socket::flush(); } - fclose($pipe); exit(0); } @@ -449,7 +308,6 @@ class Q_WebSocket // ── PARENT ── fclose($pair[1]); stream_set_blocking($pair[0], false); - $ipcWatcher = Q_Evented::onReadable($pair[0], function ($pipe) use ($socketKey) { $data = @fread($pipe, 65536); if ($data === false || $data === '') { @@ -463,74 +321,321 @@ class Q_WebSocket if ($cmd) Q_WebSocket::executeCommand($cmd); } }); - self::$workers[$socketKey] = array( - 'pid' => $pid, - 'pipe' => $pair[0], - 'watcher' => $ipcWatcher, + 'pid' => $pid, 'pipe' => $pair[0], 'watcher' => $ipcWatcher, ); } - /** - * Clean up a worker process for a socket. - * @method cleanupWorker - * @static - */ static function cleanupWorker($socketKey) { if (!isset(self::$workers[$socketKey])) return; $w = self::$workers[$socketKey]; if ($w['watcher']) Q_Evented::cancel($w['watcher']); @fclose($w['pipe']); - if ($w['pid'] > 0) { - if (function_exists("posix_kill")) posix_kill($w["pid"], SIGTERM); + if ($w['pid'] > 0 && function_exists('posix_kill')) { + posix_kill($w['pid'], SIGTERM); pcntl_waitpid($w['pid'], $st, WNOHANG); } unset(self::$workers[$socketKey]); } - /** - * Notify a worker that its WebSocket client disconnected. - * Sends a _disconnect event then cleans up. - * @method notifyDisconnect - * @static - */ static function notifyDisconnect($socketKey) { if (!isset(self::$workers[$socketKey])) return; - // Send disconnect message to child (it will exit its listen loop) $json = json_encode(array('event' => '_disconnect', 'data' => array())); $packet = pack('N', strlen($json)) . $json; @fwrite(self::$workers[$socketKey]['pipe'], $packet); - // Give child a moment then cleanup self::cleanupWorker($socketKey); } + // ── Room workers (process-per-room) ───────────── + /** - * Run a socket event handler in-process (Windows/no fork fallback). - * @method dispatchEventInProcess + * Get room patterns from config. Cached. + * Config format: + * Q.webserver.sockets.rooms.$pattern = {handler, tick?} + * e.g. "game/$id" => {"handler": "game/room", "tick": 100} + * @method getRoomPatterns * @static */ + static function getRoomPatterns() + { + if (self::$roomPatterns !== null) return self::$roomPatterns; + self::$roomPatterns = Q_Config::get('Q', 'webserver', 'sockets', 'rooms', array()); + return self::$roomPatterns; + } + + /** + * Check if a room name matches a configured room pattern. + * Returns the config (handler, tick) or null. + * @method matchRoomPattern + * @static + */ + static function matchRoomPattern($roomName) + { + $patterns = self::getRoomPatterns(); + if (empty($patterns)) return null; + $segments = explode('/', $roomName); + foreach ($patterns as $pattern => $config) { + $pSegments = explode('/', $pattern); + if (count($pSegments) !== count($segments)) continue; + $match = true; + $params = array(); + for ($i = 0; $i < count($pSegments); $i++) { + $ps = $pSegments[$i]; + if (isset($ps[0]) && ($ps[0] === '$' || $ps[0] === ':')) { + $params[substr($ps, 1)] = $segments[$i]; + } elseif ($ps !== $segments[$i]) { + $match = false; + break; + } + } + if ($match) { + return array_merge((array) $config, array('_params' => $params, '_pattern' => $pattern)); + } + } + return null; + } + + /** + * Spawn a room worker process. + * @method spawnRoomWorker + * @static + */ + static function spawnRoomWorker($roomName, $config) + { + if (!function_exists('pcntl_fork')) return; + if (isset(self::$roomWorkers[$roomName])) return; + + $pf = defined('STREAM_PF_UNIX') ? STREAM_PF_UNIX : STREAM_PF_INET; + $pair = stream_socket_pair($pf, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + if (!$pair) return; + + $handler = $config['handler'] ?? ''; + $tick = isset($config['tick']) ? (int) $config['tick'] : 0; + $params = $config['_params'] ?? array(); + + $pid = pcntl_fork(); + if ($pid === -1) { fclose($pair[0]); fclose($pair[1]); return; } + + if ($pid === 0) { + // ── CHILD: room message loop ── + fclose($pair[0]); + $pipe = $pair[1]; + Q_Socket::$_pipe = $pipe; + Q_Socket::$_socketId = 0; // room process, no single socket + + // Set up tick timer if configured + $tickCallback = null; + if ($tick > 0) { + $tickCallback = function () use ($handler, $roomName, $params, $pipe) { + Q_Socket::$_ack = null; + $result = null; + $p = array_merge($params, array( + '_room' => $roomName, 'event' => '_tick', + 'data' => array(), '_socketId' => 0, + )); + Q::event($handler, $p, false, false, $result); + Q_Socket::flush(); + }; + } + + // Fire _init event + $result = null; + Q::event($handler, array_merge($params, array( + '_room' => $roomName, 'event' => '_init', 'data' => array(), + '_socketId' => 0, + )), false, false, $result); + Q_Socket::flush(); + + // Message loop with optional tick + stream_set_blocking($pipe, false); + $lastTick = microtime(true); + + while (true) { + $read = array($pipe); + $write = $except = null; + $timeout = $tick > 0 ? max(0.001, ($tick / 1000.0) - (microtime(true) - $lastTick)) : 1.0; + $ready = @stream_select($read, $write, $except, (int) $timeout, + (int) (($timeout - (int) $timeout) * 1000000)); + + // Tick + if ($tick > 0 && (microtime(true) - $lastTick) * 1000 >= $tick) { + $lastTick = microtime(true); + if ($tickCallback) $tickCallback(); + } + + if ($ready === false) break; + if ($ready === 0) continue; + + // Read length-prefixed messages + $raw = @fread($pipe, 65536); + if ($raw === false || $raw === '') break; + + // May contain multiple messages + $buf = $raw; + while (strlen($buf) >= 4) { + $len = unpack('N', substr($buf, 0, 4))[1]; + if ($len <= 0 || $len > 10485760) { $buf = ''; break; } + if (strlen($buf) < 4 + $len) break; + $json = substr($buf, 4, $len); + $buf = substr($buf, 4 + $len); + + $msg = json_decode($json, true); + if (!$msg) continue; + + $event = $msg['event'] ?? ''; + if ($event === '_shutdown') break 2; + + Q_Socket::$_ack = isset($msg['ack']) ? $msg['ack'] : null; + Q_Socket::$_socketId = $msg['_socketId'] ?? 0; + + $result = null; + $p = array_merge($params, array( + '_room' => $roomName, + '_socketId' => Q_Socket::$_socketId, + '_ack' => Q_Socket::$_ack, + 'event' => $event, + 'data' => $msg['data'] ?? array(), + )); + Q::event($handler, $p, false, false, $result); + + if (Q_Socket::$_ack !== null && $result !== null) { + Q_Socket::send(Q_Socket::$_socketId, + array('ack' => Q_Socket::$_ack, 'data' => $result)); + } + Q_Socket::flush(); + } + } + + // Fire _destroy event + Q::event($handler, array_merge($params, array( + '_room' => $roomName, 'event' => '_destroy', 'data' => array(), + '_socketId' => 0, + )), false, false, $result); + Q_Socket::flush(); + + fclose($pipe); + exit(0); + } + + // ── PARENT ── + fclose($pair[1]); + stream_set_blocking($pair[0], false); + $ipcWatcher = Q_Evented::onReadable($pair[0], function ($pipe) use ($roomName) { + $data = @fread($pipe, 65536); + if ($data === false || $data === '') { + Q_WebSocket::cleanupRoomWorker($roomName); + return; + } + $lines = explode("\n", trim($data)); + foreach ($lines as $line) { + if ($line === '') continue; + $cmd = json_decode($line, true); + if ($cmd) Q_WebSocket::executeCommand($cmd); + } + }); + self::$roomWorkers[$roomName] = array( + 'pid' => $pid, 'pipe' => $pair[0], 'watcher' => $ipcWatcher, + 'members' => array(), + ); + } + + /** + * Send a message to a room worker. + * @method sendToRoomWorker + * @static + */ + static function sendToRoomWorker($roomName, $msg) + { + if (!isset(self::$roomWorkers[$roomName])) return; + $json = json_encode($msg, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE); + $packet = pack('N', strlen($json)) . $json; + @fwrite(self::$roomWorkers[$roomName]['pipe'], $packet); + } + + /** + * Notify room worker when a socket joins. + * @method notifyRoomJoin + * @static + */ + static function notifyRoomJoin($channel, $socketKey) + { + $config = self::matchRoomPattern($channel); + if (!$config) return; + + // Spawn room worker if not running + if (!isset(self::$roomWorkers[$channel])) { + self::spawnRoomWorker($channel, $config); + } + if (!isset(self::$roomWorkers[$channel])) return; + + self::$roomWorkers[$channel]['members'][$socketKey] = true; + self::sendToRoomWorker($channel, array( + 'event' => '_join', 'data' => array(), + '_socketId' => $socketKey, + )); + } + + /** + * Notify room worker when a socket leaves. + * @method notifyRoomLeave + * @static + */ + static function notifyRoomLeave($channel, $socketKey) + { + if (!isset(self::$roomWorkers[$channel])) return; + unset(self::$roomWorkers[$channel]['members'][$socketKey]); + + self::sendToRoomWorker($channel, array( + 'event' => '_leave', 'data' => array(), + '_socketId' => $socketKey, + )); + + // Shut down room if empty + if (empty(self::$roomWorkers[$channel]['members'])) { + self::sendToRoomWorker($channel, array( + 'event' => '_shutdown', 'data' => array(), + )); + self::cleanupRoomWorker($channel); + } + } + + /** + * Clean up a room worker. + * @method cleanupRoomWorker + * @static + */ + static function cleanupRoomWorker($roomName) + { + if (!isset(self::$roomWorkers[$roomName])) return; + $w = self::$roomWorkers[$roomName]; + if ($w['watcher']) Q_Evented::cancel($w['watcher']); + @fclose($w['pipe']); + if ($w['pid'] > 0 && function_exists('posix_kill')) { + posix_kill($w['pid'], SIGTERM); + pcntl_waitpid($w['pid'], $st, WNOHANG); + } + unset(self::$roomWorkers[$roomName]); + } + + // ── In-process fallback (Windows) ─────────────── + static function dispatchEventInProcess($eventName, $params, $socketKey, $ack) { Q_Socket::$_directMode = true; Q_Socket::$_socketId = $socketKey; Q_Socket::$_ack = $ack; - $result = null; Q::event($eventName, $params, false, false, $result); - if ($ack !== null && $result !== null) { self::send($socketKey, array('ack' => $ack, 'data' => $result)); } Q_Socket::$_directMode = false; } - /** - * Execute an IPC command from a child process. - * @method executeCommand - * @static - */ + // ── IPC command execution ─────────────────────── + static function executeCommand($cmd) { switch ($cmd['cmd'] ?? '') { diff --git a/web/qbix-socket.js b/web/qbix-socket.js index 708775d..b23f982 100644 --- a/web/qbix-socket.js +++ b/web/qbix-socket.js @@ -1,23 +1,32 @@ /** - * QSocket β€” tiny WebSocket client for Qbix Server. + * QSocket β€” WebSocket client for Qbix Server. * * Usage: * var qs = new QSocket('ws://localhost:8080/ws/chat'); * + * // Listen for events from the server * qs.on('chat/message', function(data) { - * console.log(data.from + ': ' + data.text); + * console.log(data.user + ': ' + data.text); * }); * - * qs.emit('chat/message', {text: 'hello'}, function(ack) { - * console.log('Server confirmed:', ack); + * // Send event with callback (server acks with structured data) + * qs.emit('chat/message', {text: 'hello'}, function(response) { + * // response is whatever the PHP handler set as $result + * // arrays, objects, nested structures β€” all preserved via JSON + * console.log('Message #' + response.count); * }); * - * qs.emit('chat/join', {room: 'lobby'}); + * // Send without callback + * qs.emit('chat/typing', {user: 'Alice'}); * * Protocol (JSON over WebSocket): * Client β†’ Server: {"event": "...", "data": {...}, "ack": N} * Server β†’ Client: {"event": "...", "data": {...}} (broadcast) * Server β†’ Client: {"ack": N, "data": {...}} (callback) + * + * Data is serialized as JSON in both directions. PHP arrays and nested + * objects map to JS objects/arrays. Callbacks receive the full structured + * response β€” strings, numbers, booleans, arrays, nested objects. */ (function (root) { 'use strict'; @@ -39,38 +48,31 @@ self.ws.onopen = function () { self._delay = self._reconnectDelay; - // Flush queued messages while (self._queue.length) { self.ws.send(self._queue.shift()); } - if (self._handlers['connect']) { - self._handlers['connect'].forEach(function (fn) { fn(); }); - } + self._fire('connect'); }; self.ws.onmessage = function (e) { var msg; try { msg = JSON.parse(e.data); } catch (err) { return; } - // Ack response (callback from server) + // Ack response β€” invoke the stored callback with full data if (msg.ack !== undefined && self._acks[msg.ack]) { self._acks[msg.ack](msg.data); delete self._acks[msg.ack]; return; } - // Event broadcast - if (msg.event && self._handlers[msg.event]) { - self._handlers[msg.event].forEach(function (fn) { - fn(msg.data); - }); + // Event broadcast β€” pass full data to all listeners + if (msg.event) { + self._fire(msg.event, msg.data); } }; self.ws.onclose = function () { - if (self._handlers['disconnect']) { - self._handlers['disconnect'].forEach(function (fn) { fn(); }); - } + self._fire('disconnect'); if (self._reconnect) { self._delay = Math.min(self._delay * 1.5, self._maxDelay); setTimeout(self._connect, self._delay); @@ -82,13 +84,22 @@ }; }; + self._fire = function (event, data) { + var handlers = self._handlers[event]; + if (!handlers) return; + for (var i = 0; i < handlers.length; i++) { + handlers[i](data); + } + }; + self._delay = self._reconnectDelay; self._connect(); } /** * Listen for an event from the server. - * Special events: 'connect', 'disconnect' + * Callback receives the data object (arrays, nested objects preserved). + * Special events: 'connect' (no data), 'disconnect' (no data) */ QSocket.prototype.on = function (event, fn) { if (!this._handlers[event]) this._handlers[event] = []; @@ -97,21 +108,31 @@ }; /** - * Remove a listener. + * Remove listener(s). No fn = remove all for that event. */ QSocket.prototype.off = function (event, fn) { if (!this._handlers[event]) return this; if (!fn) { delete this._handlers[event]; return this; } - this._handlers[event] = this._handlers[event].filter(function (f) { return f !== fn; }); + this._handlers[event] = this._handlers[event].filter(function (f) { + return f !== fn; + }); return this; }; /** * Send an event to the server. - * Optional callback is invoked when the server acks. + * + * @param {string} event - Event name (maps to PHP handler) + * @param {*} data - Any JSON-serializable value: object, array, string, number, boolean, null + * @param {function} [callback] - Called with the server's response (the PHP handler's $result) + * + * Examples: + * qs.emit('chat/message', {text: 'hi', tags: ['urgent']}, function(res) { ... }); + * qs.emit('game/move', {x: 10, y: 20}); + * qs.emit('ping', null, function(res) { console.log(res.time); }); */ QSocket.prototype.emit = function (event, data, callback) { - var msg = { event: event, data: data || {} }; + var msg = { event: event, data: (data !== undefined ? data : null) }; if (typeof callback === 'function') { msg.ack = ++this._ackId; this._acks[msg.ack] = callback; @@ -126,7 +147,7 @@ }; /** - * Close the connection (disables auto-reconnect). + * Close the connection. Disables auto-reconnect. */ QSocket.prototype.close = function () { this._reconnect = false;