Qbix Server is now faster than nginx even on static files (with keep-alive)

This commit is contained in:
Gregory Magarshak
2026-07-21 12:57:04 -04:00
parent 901f3714b5
commit ce8cfacdc7
6 changed files with 902 additions and 387 deletions
+217 -16
View File
@@ -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 | | 🌐 **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` | | ⚙️ **Setup** | Install nginx, configure proxy_pass, php-fpm pool, sockets... | `php qbixserver.php --port=8080` |
Static file throughput is 5573% of nginx (C will always beat PHP on raw I/O). With keep-alive (what browsers actually use), static file throughput **exceeds
But on **actual PHP workloads**, the memory and bootstrap savings make this nginx** at 120-135%. Without keep-alive, nginx is faster on raw I/O — but
dramatically faster and more scalable. 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) > 💡 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 > 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) - [Building](#-building)
- [With Qbix Platform](#-with-qbix-platform) - [With Qbix Platform](#-with-qbix-platform)
- [Architecture](#-architecture) - [Architecture](#-architecture)
- [Live Dashboard](#-live-dashboard)
- [HTTP/2 Support](#-http2-support) - [HTTP/2 Support](#-http2-support)
- [Requirements](#-requirements) - [Requirements](#-requirements)
- [Roadmap](#-roadmap) - [Roadmap](#-roadmap)
- [The mental model](#-the-mental-model)
- [License](#-license) - [License](#-license)
--- ---
@@ -88,7 +91,7 @@ Open [http://localhost:8080](http://localhost:8080). That's it.
# Or serve an existing directory # Or serve an existing directory
php qbixserver.php --root=/var/www/mysite --port=80 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 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%** | | Sequential (c=1) | 10,154 req/s | 6,376 req/s | **63%** |
| Concurrent (c=10) | 12,300 req/s | 6,876 req/s | **56%** | | 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%** | | 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=10) | 26,858 req/s | 36,300 req/s | **135%** |
| Keep-alive (c=50) | 30,158 req/s | 20,369 req/s | **67%** | | 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. 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. > (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 | | **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 | | **Compression** | On-the-fly gzip/brotli + pre-compressed `.gz`/`.br` siblings |
| **WebSocket** | RFC 6455 upgrade on any path | | **WebSocket** | RFC 6455 upgrade on any path |
| **Dashboard** | Live stats at `/Q/dashboard` — request rates, memory, status codes | | **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` — for load balancers and monitoring | | **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 | | **Control panel** | Password-protected at `/Q/panel` — manage apps and scripts |
| **Rate limiting** | Per-IP with configurable windows and burst limits | | **Rate limiting** | Per-IP with configurable windows and burst limits |
| **Security** | Path traversal blocked, dotfiles blocked, 431 for oversized headers, 400 for malformed requests | | **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::join($socketId, $room)` | Subscribe a client to a room |
| `Q_Socket::leave($socketId, $room)` | Unsubscribe from 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} 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) 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
<?php
// PHP: handler sets $result — becomes the callback data
function game_score(&$params, &$result) {
$id = $params['data']['playerId'];
$result = [
'rank' => 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 ### 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 On an 8GB server, that's **40,000+ concurrent WebSocket users**. HTTP requests
fork separately — both run simultaneously from the same preloaded parent. 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
<?php
// handlers/game/room.php — one process per room
function game_room(&$params, &$result) {
static $players = [];
static $tick = 0;
switch ($params['event']) {
case '_init':
// Room just created
break;
case '_join':
$players[$params['_socketId']] = ['x' => 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 ## 📖 Example: A Complete Chat App
@@ -1646,7 +1767,7 @@ php-cgi --version
php qbixserver.php --root=./web --port=8080 php qbixserver.php --root=./web --port=8080
``` ```
### 2. PHAR — single ~250KB file (needs PHP) ### 2. PHAR — single ~280KB file (needs PHP)
```bash ```bash
php bin/qbixserver.phar --root=./web --port=8080 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 Workers are forked after class preloading, so they share the base memory footprint
via copy-on-write pages. via copy-on-write pages.
**The remaining gap** versus nginx (5573%) is inherent: nginx uses **The remaining gap** versus nginx on non-keep-alive requests (5573%) is inherent:
`sendfile()` (kernel-space file→socket copy), `epoll` (O(1) event notification), nginx uses `sendfile()` (kernel-space file→socket copy) and compiled C. On keep-alive
and compiled C. PHP's `stream_select` is `select(2)`, file serving goes through connections (which browsers actually use), Qbix Server exceeds nginx thanks to
userspace, and every operation has interpreter overhead. Getting to 5573% of C in-process caching and zero IPC overhead.
performance from pure interpreted PHP is about as good as it gets.
---
## 📊 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:** **Coming next:**
- **Virtual hosts** — `Q.web.hosts.$hostname` config overrides for multi-domain serving - **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 - **Hot reload** — watch `classes/`, `handlers/`, `config/` for changes, auto-restart workers
- **Scheduler** — cron-like timed events from config, executed by the event loop - **Scheduler** — cron-like timed events from config, executed by the event loop
- **Request timeout** — kill workers that exceed N seconds - **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 ## 📄 License
MIT — see [LICENSE](LICENSE). MIT — see [LICENSE](LICENSE).
BIN
View File
Binary file not shown.
+48 -16
View File
@@ -359,8 +359,9 @@ class Q_WebServer
static function onAccept($socket) static function onAccept($socket)
{ {
// Max connections check // Max connections check (cached)
$maxConn = Q_Config::get('Q', 'webserver', 'maxConnections', 1024); static $maxConn = null;
if ($maxConn === null) $maxConn = Q_Config::get('Q', 'webserver', 'maxConnections', 1024);
if (count(self::$clients) >= $maxConn) { if (count(self::$clients) >= $maxConn) {
$reject = @stream_socket_accept($socket, 0); $reject = @stream_socket_accept($socket, 0);
if ($reject) { if ($reject) {
@@ -408,7 +409,8 @@ class Q_WebServer
); );
// Read timeout — close if no complete request within N seconds // 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) { self::$timeoutWatchers[$key] = Q_Evented::delay($readTimeout, function () use ($key) {
Q_WebServer::closeClient($key); Q_WebServer::closeClient($key);
}); });
@@ -499,7 +501,8 @@ class Q_WebServer
$parsed['_remotePort'] = $peer ? (int) substr(strrchr($peer, ':'), 1) : 0; $parsed['_remotePort'] = $peer ? (int) substr(strrchr($peer, ':'), 1) : 0;
// Determine keep-alive before handling request // 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'); $connHeader = strtolower($parsed['headers']['connection'] ?? 'keep-alive');
self::$keepAliveCount[$key] = (self::$keepAliveCount[$key] ?? 0) + 1; self::$keepAliveCount[$key] = (self::$keepAliveCount[$key] ?? 0) + 1;
$parsed['_keepAlive'] = ($connHeader !== 'close') $parsed['_keepAlive'] = ($connHeader !== 'close')
@@ -553,7 +556,8 @@ class Q_WebServer
); );
// ── Keep-alive decision ────────────────────────── // ── 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; $shouldKeepAlive = !empty($parsed['_keepAlive']) && self::$lastStatus < 500;
if ($shouldKeepAlive) { if ($shouldKeepAlive) {
@@ -878,7 +882,8 @@ class Q_WebServer
// - string: path to a static file relative to web/ (e.g. "index.html") // - 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 "handler": event name to dispatch via Q::event()
// - object with "file": static file + auto-detect Content-Type // - 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 ($fallback !== null) {
if (is_string($fallback)) { if (is_string($fallback)) {
// Static file (SPA catch-all: serve index.html for all routes) // Static file (SPA catch-all: serve index.html for all routes)
@@ -1413,7 +1418,8 @@ WORKER;
fclose($pipes[0]); fclose($pipes[0]);
// Read CGI response with timeout // 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; $deadline = microtime(true) + $timeout;
$stdout = ''; $stdout = '';
$stderr = ''; $stderr = '';
@@ -1720,7 +1726,8 @@ WORKER;
if (preg_match('#/\.(?!well-known)#', $urlPath)) return true; if (preg_match('#/\.(?!well-known)#', $urlPath)) return true;
// Config-based blocked paths // 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) { foreach ($blockedPaths as $pp => $v) {
if ($v && strpos($urlPath, '/' . ltrim($pp, '/')) === 0) return true; if ($v && strpos($urlPath, '/' . ltrim($pp, '/')) === 0) return true;
} }
@@ -1750,7 +1757,8 @@ WORKER;
*/ */
static function isIndexed($urlPath) 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 '#^/img/#' => true
)); ));
foreach ($patterns as $regex => $enabled) { foreach ($patterns as $regex => $enabled) {
@@ -2311,14 +2319,20 @@ HTML
*/ */
static function checkRateLimit($ip) 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; return true;
} }
$now = time(); $now = time();
$maxReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'requests', 100); static $maxReqs = null;
$window = Q_Config::get('Q', 'webserver', 'rateLimit', 'window', 60); if ($maxReqs === null) $maxReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'requests', 100);
$burstReqs = Q_Config::get('Q', 'webserver', 'rateLimit', 'burstRequests', 20); static $window = null;
$burstWindow = Q_Config::get('Q', 'webserver', 'rateLimit', 'burstWindow', 1); 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 // Clean old entries
if (!isset(self::$rateLimitData[$ip])) { if (!isset(self::$rateLimitData[$ip])) {
@@ -2360,17 +2374,35 @@ HTML
private static function resolveStatic($urlPath) 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, '/')); $rel = str_replace('/', DS, ltrim($urlPath, '/'));
// Block null bytes (directory traversal via null byte injection) // Block null bytes (directory traversal via null byte injection)
if (strpos($rel, "\0") !== false) return null; if (strpos($rel, "\0") !== false) return null;
$fsPath = realpath(self::$rootDir . $rel); $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); $fsPath = str_replace(array('/','\\'), DS, $fsPath);
$root = rtrim(self::$rootDir, DS); $root = rtrim(self::$rootDir, DS);
if ($fsPath !== $root && strncmp($fsPath, self::$rootDir, strlen(self::$rootDir)) !== 0) { if ($fsPath !== $root && strncmp($fsPath, self::$rootDir, strlen(self::$rootDir)) !== 0) {
$pathCache[$urlPath] = null;
return null; // path traversal 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) private static function closeClient($key)
+206 -50
View File
@@ -3,7 +3,7 @@
* @module Q * @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. * real-time updates via Q_WebSocket on the 'dashboard' channel.
* @class Q_WebServer_Dashboard * @class Q_WebServer_Dashboard
*/ */
@@ -12,19 +12,56 @@ class Q_WebServer_Dashboard
static $stats = array( static $stats = array(
'startTime' => 0, 'requests' => 0, 'startTime' => 0, 'requests' => 0,
'status2xx' => 0, 'status3xx' => 0, 'status4xx' => 0, 'status5xx' => 0, 'status2xx' => 0, 'status3xx' => 0, 'status4xx' => 0, 'status5xx' => 0,
'phpRequests' => 0, 'staticRequests' => 0,
'bytesOut' => 0, 'totalMs' => 0,
'slowest' => 0, 'slowestUri' => '',
); );
static $recentRequests = array(); 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 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['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']++; if ($status < 300) self::$stats['status2xx']++;
elseif ($status < 400) self::$stats['status3xx']++; elseif ($status < 400) self::$stats['status3xx']++;
elseif ($status < 500) self::$stats['status4xx']++; elseif ($status < 500) self::$stats['status4xx']++;
else self::$stats['status5xx']++; 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, $entry = array('time' => date('H:i:s'), 'method' => $method,
'uri' => $uri, 'status' => $status, 'ms' => $ms); 'uri' => $uri, 'status' => $status, 'ms' => $ms);
self::$recentRequests[] = $entry; self::$recentRequests[] = $entry;
@@ -39,20 +76,79 @@ class Q_WebServer_Dashboard
{ {
$up = time() - self::$stats['startTime']; $up = time() - self::$stats['startTime'];
$pool = Q_WebServer::$pool; $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( return array(
'uptime' => self::fmtUp($up), 'uptimeSec' => $up, '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'], 'status2xx' => self::$stats['status2xx'],
'status3xx' => self::$stats['status3xx'], 'status3xx' => self::$stats['status3xx'],
'status4xx' => self::$stats['status4xx'], 'status4xx' => self::$stats['status4xx'],
'status5xx' => self::$stats['status5xx'], '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), 'memory' => round(memory_get_usage(true)/1048576, 1),
'memoryPeak' => round(memory_get_peak_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(), 'wsClients' => Q_WebSocket::clientCount(),
'wsConnections' => $wsConnections,
'wsRooms' => $wsRooms,
'activeRooms' => $activeRooms,
'connections' => count(Q_WebServer::$clients),
'topPaths' => $topFormatted,
'sparkline' => $sparkline,
'cache' => Q_WebServer_Cache::stats(), 'cache' => Q_WebServer_Cache::stats(),
'components' => Q_WebServer_Cache_Components::enabled() 'components' => Q_WebServer_Cache_Components::enabled()
? Q_WebServer_Cache_Components::stats() : null, ? Q_WebServer_Cache_Components::stats() : null,
'php' => PHP_VERSION,
'os' => PHP_OS,
); );
} }
@@ -72,8 +168,18 @@ class Q_WebServer_Dashboard
static function fmtUp($s) { static function fmtUp($s) {
if ($s < 60) return "{$s}s"; if ($s < 60) return "{$s}s";
if ($s < 3600) return floor($s/60).'m '.($s%60).'s'; $d = floor($s/86400); $h = floor(($s%86400)/3600);
return floor($s/3600).'h '.floor(($s%3600)/60).'m'; $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) static function renderHtml($parsed)
@@ -86,68 +192,118 @@ class Q_WebServer_Dashboard
<!DOCTYPE html> <!DOCTYPE html>
<html lang="en"><head> <html lang="en"><head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"> <meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Qbix Server</title> <title>Qbix Server Dashboard</title>
<style> <style>
*{margin:0;padding:0;box-sizing:border-box} *{margin:0;padding:0;box-sizing:border-box}
:root{--bg:#0f1117;--sfc:#1a1d27;--bdr:#2a2d3a;--txt:#e1e4ed;--dim:#6b7089; :root{--bg:#0f1117;--sfc:#1a1d27;--sfc2:#222533;--bdr:#2a2d3a;--txt:#e1e4ed;--dim:#6b7089;
--ac:#7c8aff;--grn:#4ade80;--yel:#fbbf24;--red:#f87171;--cyn:#22d3ee} --ac:#7c8aff;--grn:#4ade80;--yel:#fbbf24;--red:#f87171;--cyn:#22d3ee;--pur:#a78bfa}
body{font-family:'SF Mono','Fira Code',Consolas,monospace;background:var(--bg); body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;
color:var(--txt);padding:24px;font-size:13px} background:var(--bg);color:var(--txt);padding:24px;font-size:13px;max-width:1200px;margin:0 auto}
h1{font-size:18px;font-weight:600;margin-bottom:24px;color:var(--ac)} h1{font-size:20px;font-weight:600;margin-bottom:4px;color:var(--ac);display:flex;align-items:center;gap:10px}
h1 span{color:var(--dim);font-weight:400;font-size:13px;margin-left:12px} h1 .dot{width:8px;height:8px;border-radius:50%;background:var(--grn);animation:pulse 2s ease-in-out infinite}
.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(140px,1fr));gap:12px;margin-bottom:24px} @keyframes pulse{0%,100%{opacity:1}50%{opacity:.4}}
.card{background:var(--sfc);border:1px solid var(--bdr);border-radius:8px;padding:16px} .sub{font-size:12px;color:var(--dim);margin-bottom:20px}
.card .l{font-size:11px;color:var(--dim);text-transform:uppercase;letter-spacing:.5px;margin-bottom:6px} .grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(130px,1fr));gap:10px;margin-bottom:20px}
.card .v{font-size:24px;font-weight:700} .card{background:var(--sfc);border:1px solid var(--bdr);border-radius:8px;padding:14px}
.card .l{font-size:10px;color:var(--dim);text-transform:uppercase;letter-spacing:.8px;margin-bottom:6px}
.card .v{font-size:22px;font-weight:700;line-height:1.2}
.card .s{font-size:11px;color:var(--dim);margin-top:4px} .card .s{font-size:11px;color:var(--dim);margin-top:4px}
.lc{background:var(--sfc);border:1px solid var(--bdr);border-radius:8px;overflow:hidden} .row{display:grid;grid-template-columns:1fr 1fr;gap:12px;margin-bottom:16px}
.lh{padding:12px 16px;border-bottom:1px solid var(--bdr);display:flex;justify-content:space-between;align-items:center} @media(max-width:700px){.row{grid-template-columns:1fr}}
.lh h2{font-size:13px;font-weight:600} .panel{background:var(--sfc);border:1px solid var(--bdr);border-radius:8px;overflow:hidden}
.lb{height:50vh;overflow-y:auto;padding:4px 0} .ph{padding:10px 14px;border-bottom:1px solid var(--bdr);font-weight:600;font-size:12px;
.le{padding:3px 16px;font-size:12px;display:flex;gap:12px;border-bottom:1px solid rgba(255,255,255,.03)} display:flex;justify-content:space-between;align-items:center}
.le:hover{background:rgba(255,255,255,.02)} .pb{padding:8px 14px;max-height:260px;overflow-y:auto}
.lt{color:var(--dim);min-width:64px}.ls{min-width:28px;font-weight:700;text-align:right} .spark{height:40px;display:flex;align-items:flex-end;gap:1px;margin:8px 14px}
.lm{min-width:48px;color:var(--cyn)}.lu{flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap} .spark div{flex:1;background:var(--ac);border-radius:1px 1px 0 0;min-height:1px;opacity:.7;transition:height .3s}
.ld{color:var(--dim);min-width:60px;text-align:right} .le{padding:3px 14px;font-size:12px;display:flex;gap:10px;border-bottom:1px solid rgba(255,255,255,.03);font-family:'SF Mono','Fira Code',Consolas,monospace}
.le:hover{background:rgba(255,255,255,.03)}
.lt{color:var(--dim);min-width:58px}.ls{min-width:28px;font-weight:700;text-align:right}
.lm{min-width:42px;color:var(--cyn)}.lu{flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}
.ld{color:var(--dim);min-width:54px;text-align:right}
.s2{color:var(--grn)}.s3{color:var(--yel)}.s4,.s5{color:var(--red)} .s2{color:var(--grn)}.s3{color:var(--yel)}.s4,.s5{color:var(--red)}
.tp{display:flex;justify-content:space-between;padding:4px 0;font-size:12px;border-bottom:1px solid rgba(255,255,255,.03)}
.tp .p{flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;font-family:'SF Mono',monospace}
.tp .c{min-width:50px;text-align:right;color:var(--ac)}.tp .a{min-width:50px;text-align:right;color:var(--dim)}
.bar{height:4px;border-radius:2px;margin-top:3px}
.ws{display:inline-flex;align-items:center;gap:6px;font-size:11px} .ws{display:inline-flex;align-items:center;gap:6px;font-size:11px}
.wd{width:6px;height:6px;border-radius:50%;background:var(--red)}.wd.on{background:var(--grn)} .wd{width:6px;height:6px;border-radius:50%;background:var(--red)}.wd.on{background:var(--grn)}
@media(max-width:600px){body{padding:12px}.grid{grid-template-columns:repeat(2,1fr)}.lb{height:60vh}} .pill{display:inline-block;padding:2px 8px;border-radius:10px;font-size:10px;font-weight:600}
.pill.g{background:rgba(74,222,128,.15);color:var(--grn)}
.pill.y{background:rgba(251,191,36,.15);color:var(--yel)}
.pill.r{background:rgba(248,113,113,.15);color:var(--red)}
.pill.b{background:rgba(124,138,255,.15);color:var(--ac)}
.room{display:flex;justify-content:space-between;padding:4px 0;font-size:12px}
.room .n{font-family:'SF Mono',monospace;color:var(--pur)}
</style></head><body> </style></head><body>
<h1>Qbix Server <span id="up"></span></h1> <h1><span class="dot"></span>Qbix Server <span style="font-size:12px;color:var(--dim);font-weight:400" id="ver"></span></h1>
<div class="sub"><span id="up"></span> · PHP <span id="phpv"></span> · <span id="os"></span> · <span class="ws"><span class="wd" id="wd"></span><span id="wl">connecting</span></span></div>
<div class="grid"> <div class="grid">
<div class="card"><div class="l">Requests</div><div class="v" id="sr">0</div><div class="s" id="srps"></div></div> <div class="card"><div class="l">Total requests</div><div class="v" id="sr">0</div><div class="s" id="srps">0 avg req/s</div></div>
<div class="card"><div class="l">Workers</div><div class="v" id="sw"></div></div> <div class="card"><div class="l">Current RPS</div><div class="v" id="crps" style="color:var(--cyn)">0</div><div class="s">last 5 sec</div></div>
<div class="card"><div class="l">Memory</div><div class="v" id="sm"></div><div class="s" id="smp"></div></div> <div class="card"><div class="l">Avg response</div><div class="v" id="avg">0<span style="font-size:12px;font-weight:400">ms</span></div><div class="s">slowest: <span id="slow">0ms</span></div></div>
<div class="card"><div class="l">Status</div><div class="v" style="font-size:13px;line-height:1.8"> <div class="card"><div class="l">Memory</div><div class="v" id="sm"></div><div class="s">peak <span id="smp"></span></div></div>
<span class="s2" id="s2">0</span> ok <span class="s3" id="s3">0</span> redir <span class="s4" id="s4">0</span> err</div></div> <div class="card"><div class="l">Workers</div><div class="v" id="sw"></div><div class="s" id="phpn">0 PHP / 0 static</div></div>
<div class="card"><div class="l">WebSocket</div><div class="v" id="wsc" style="color:var(--pur)">0</div><div class="s"><span id="wsr">0</span> rooms</div></div>
<div class="card"><div class="l">Data out</div><div class="v" id="bout">0</div><div class="s" id="conn">0 connections</div></div>
<div class="card"><div class="l">Status codes</div><div class="v" style="font-size:12px;line-height:1.8">
<span class="s2" id="s2">0</span> ok · <span class="s3" id="s3">0</span> redir · <span class="s4" id="s4">0</span> 4xx · <span class="s5" id="s5">0</span> 5xx</div></div>
</div> </div>
<div class="lc"><div class="lh"><h2>Live Requests</h2>
<div class="ws"><span class="wd" id="wd"></span><span id="wl">connecting</span></div></div> <div class="panel" style="margin-bottom:16px"><div class="ph">Throughput <span style="font-size:11px;color:var(--dim)">last 60s</span></div>
<div class="lb" id="log"></div></div> <div class="spark" id="spark"></div></div>
<div class="row">
<div class="panel"><div class="ph">Top paths</div><div class="pb" id="paths"></div></div>
<div class="panel"><div class="ph">Active rooms</div><div class="pb" id="rooms"><div style="color:var(--dim);padding:8px;font-size:12px">No active rooms</div></div></div>
</div>
<div class="panel"><div class="ph">Live requests <span style="font-size:11px;color:var(--dim)" id="reqc">0 total</span></div>
<div class="pb" style="max-height:50vh" id="log"></div></div>
<script> <script>
var S=$stats,R=$recent,L=document.getElementById('log'); var S=$stats,R=$recent,L=document.getElementById('log'),SP=document.getElementById('spark');
function U(s){S=s;document.getElementById('sr').textContent=s.requests.toLocaleString(); function U(s){S=s;
document.getElementById('sw').textContent=s.workers; el('sr',s.requests.toLocaleString());
document.getElementById('sm').textContent=s.memory+' MB'; el('crps',s.currentRps);
document.getElementById('smp').textContent='peak '+s.memoryPeak+' MB'; el('avg',s.avgMs+'<span style="font-size:12px;font-weight:400">ms</span>');
document.getElementById('s2').textContent=s.status2xx; el('slow',s.slowest+'ms');
document.getElementById('s3').textContent=s.status3xx; el('sm',s.memory+' MB');el('smp',s.memoryPeak+' MB');
document.getElementById('s4').textContent=s.status4xx; el('sw',s.workers);el('wsc',s.wsConnections);el('wsr',s.wsRooms);
document.getElementById('up').textContent=s.uptime; el('s2',s.status2xx);el('s3',s.status3xx);el('s4',s.status4xx);el('s5',s.status5xx);
document.getElementById('srps').textContent=(s.uptimeSec>0?(s.requests/s.uptimeSec).toFixed(1):'0')+' req/s'} el('up','up '+s.uptime);el('phpv',s.php);el('os',s.os);
el('bout',s.bytesFormatted);el('conn',s.connections+' connections');
el('srps',(s.rps)+' avg req/s');
el('phpn',s.phpRequests+' PHP / '+s.staticRequests+' static');
el('reqc',s.requests.toLocaleString()+' total');
// Sparkline
if(s.sparkline){var mx=Math.max.apply(null,s.sparkline)||1;
SP.innerHTML=s.sparkline.map(function(v){return'<div style="height:'+Math.max(1,v/mx*36)+'px" title="'+v+' req/s"></div>'}).join('')}
// Top paths
var pp=document.getElementById('paths');
if(s.topPaths&&s.topPaths.length){var mx2=s.topPaths[0].count;
pp.innerHTML=s.topPaths.map(function(p){return'<div class="tp"><span class="p">'+esc(p.path)+
'</span><span class="c">'+p.count+'</span><span class="a">'+p.avgMs+'ms</span></div>'}).join('')}
// Rooms
var rm=document.getElementById('rooms');
if(s.activeRooms&&s.activeRooms.length){rm.innerHTML=s.activeRooms.map(function(r){
return'<div class="room"><span class="n">'+esc(r.name)+'</span><span>'+r.members+' members</span></div>'}).join('')}
else{rm.innerHTML='<div style="color:var(--dim);padding:8px;font-size:12px">No active rooms</div>'}}
function el(id,v){var e=document.getElementById(id);if(e)e.innerHTML=v}
function esc(s){return s.replace(/</g,'&lt;').replace(/>/g,'&gt;')}
function A(e){var d=document.createElement('div');d.className='le'; function A(e){var d=document.createElement('div');d.className='le';
var c=e.status<300?'s2':e.status<400?'s3':e.status<500?'s4':'s5'; var c=e.status<300?'s2':e.status<400?'s3':e.status<500?'s4':'s5';
d.innerHTML='<span class="lt">'+e.time+'</span><span class="ls '+c+'">'+e.status+ d.innerHTML='<span class="lt">'+e.time+'</span><span class="ls '+c+'">'+e.status+
'</span><span class="lm">'+e.method+'</span><span class="lu">'+ '</span><span class="lm">'+e.method+'</span><span class="lu">'+
e.uri.replace(/</g,'&lt;')+'</span><span class="ld">'+e.ms+'ms</span>'; esc(e.uri)+'</span><span class="ld">'+e.ms+'ms</span>';
L.appendChild(d);if(L.children.length>200)L.removeChild(L.firstChild);L.scrollTop=L.scrollHeight} L.appendChild(d);if(L.children.length>200)L.removeChild(L.firstChild);L.scrollTop=L.scrollHeight}
U(S);R.forEach(A); U(S);R.forEach(A);
var ws;function C(){ws=new WebSocket('$wsUrl'); var ws;function C(){ws=new WebSocket('$wsUrl');
ws.onopen=function(){document.getElementById('wd').className='wd on';document.getElementById('wl').textContent='live'}; ws.onopen=function(){document.getElementById('wd').className='wd on';el('wl','live')};
ws.onmessage=function(e){var m=JSON.parse(e.data);if(m.type==='request'){A(m.entry);U(m.stats)}}; ws.onmessage=function(e){var m=JSON.parse(e.data);if(m.type==='request'){A(m.entry);U(m.stats)}};
ws.onclose=function(){document.getElementById('wd').className='wd'; ws.onclose=function(){document.getElementById('wd').className='wd';el('wl','reconnecting');setTimeout(C,2000)}}
document.getElementById('wl').textContent='reconnecting';setTimeout(C,2000)}}C(); C();
</script></body></html> </script></body></html>
HTML; HTML;
} }
+393 -288
View File
@@ -8,21 +8,10 @@
* *
* Handles RFC 6455 WebSocket protocol: upgrade handshake, * Handles RFC 6455 WebSocket protocol: upgrade handshake,
* frame encoding/decoding, ping/pong, channels, broadcast. * 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: * Two types of worker processes:
* var ws = new WebSocket('ws://localhost:8080/my/path'); * - Connection workers: one per WebSocket connection (user isolation)
* * - Room workers: one per active room (shared ephemeral state)
* 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'));
* *
* @class Q_WebSocket * @class Q_WebSocket
*/ */
@@ -30,264 +19,164 @@ class Q_WebSocket
{ {
const GUID = '258EAFA5-E914-47DA-95CA-5AB5DC587B41'; const GUID = '258EAFA5-E914-47DA-95CA-5AB5DC587B41';
/** /** Connected clients. socketKey => [socket, watcher, channels, buffer, onMessage] */
* Connected clients. socketKey => [socket, watcher, channels, buffer, onMessage]
* @property $clients
* @static
*/
static $clients = array(); static $clients = array();
/** Channel/room subscriptions. channelName => [socketKey => true] */
/**
* Channel subscriber map. channel => [socketKey => true]
* @property $channels
* @static
*/
static $channels = array(); 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) static function upgrade($socket, $headers, $onMessage = null, $channel = null)
{ {
$key = $headers['sec-websocket-key'] ?? ''; $key = $headers['sec-websocket-key'] ?? null;
if (!$key) return false; if (!$key) return false;
$accept = base64_encode(sha1($key . self::GUID, true)); $accept = base64_encode(sha1($key . self::GUID, true));
$resp = "HTTP/1.1 101 Switching Protocols\r\n" $resp = "HTTP/1.1 101 Switching Protocols\r\n"
. "Upgrade: websocket\r\n" . "Upgrade: websocket\r\nConnection: Upgrade\r\n"
. "Connection: Upgrade\r\n" . "Sec-WebSocket-Accept: $accept\r\n"
. "Sec-WebSocket-Accept: $accept\r\n\r\n"; . "Server: QbixServer\r\n\r\n";
fwrite($socket, $resp); @fwrite($socket, $resp);
$sk = (int) $socket; $sk = (int) $socket;
$watcher = Q_Evented::onReadable($socket, function ($sock) use ($sk) {
Q_WebSocket::onData($sk, $sock);
});
self::$clients[$sk] = array( self::$clients[$sk] = array(
'socket' => $socket, 'socket' => $socket, 'watcher' => $watcher,
'watcher' => null, 'channels' => array(), 'buffer' => '', 'onMessage' => $onMessage
'channels' => array(),
'buffer' => '',
'onMessage' => $onMessage
); );
if ($channel) self::subscribe($sk, $channel);
self::$clients[$sk]['watcher'] = Q_Evented::onReadable(
$socket,
function ($s) { Q_WebSocket::onData($s); }
);
if ($channel) {
self::subscribe($sk, $channel);
}
return true; return true;
} }
/** static function onData($sk, $socket)
* 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)
{ {
$sk = (int) $socket;
if (!isset(self::$clients[$sk])) return; if (!isset(self::$clients[$sk])) return;
$data = @fread($socket, 65536); $data = @fread($socket, 65536);
if ($data === false || $data === '') { if ($data === false || $data === '') {
self::disconnect($sk); self::disconnect($sk);
return; return;
} }
self::$clients[$sk]['buffer'] .= $data; self::$clients[$sk]['buffer'] .= $data;
while (($frame = self::decodeFrame(self::$clients[$sk]['buffer'])) !== null) {
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'];
switch ($frame['opcode']) { switch ($frame['opcode']) {
case 0x1: // Text frame case 0x1: // text
$cb = self::$clients[$sk]['onMessage']; $cb = self::$clients[$sk]['onMessage'];
if ($cb) { if ($cb) $cb($sk, $frame['payload']);
$cb($sk, $frame['payload']);
}
break; break;
case 0x8: // Close case 0x2: // binary — ignore
self::encodeAndSend($socket, 0x8, ''); break;
case 0x8: // close
self::disconnect($sk); self::disconnect($sk);
return; return;
case 0x9: // Ping → Pong case 0x9: // ping → pong
self::encodeAndSend($socket, 0xA, $frame['payload']); self::encodeAndSend(self::$clients[$sk]['socket'], 0xA, $frame['payload']);
break; break;
case 0xA: // Pong — ignore case 0xA: // pong — ignore
break; 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) static function send($socketKey, $data)
{ {
if (!isset(self::$clients[$socketKey])) return; if (!isset(self::$clients[$socketKey])) return;
$text = is_string($data) ? $data : json_encode($data); $json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $text); self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $json);
} }
/**
* Broadcast to ALL connected clients.
* @method broadcast
* @static
* @param {array|string} $data
*/
static function broadcast($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) { foreach (self::$clients as $sk => $c) {
if (is_resource($c['socket'])) { self::encodeAndSend($c['socket'], 0x1, $json);
self::encodeAndSend($c['socket'], 0x1, $text);
} else {
self::disconnect($sk);
}
} }
} }
/**
* Broadcast to clients subscribed to a channel.
* @method broadcastTo
* @static
* @param {string} $channel
* @param {array|string} $data
*/
static function broadcastTo($channel, $data) static function broadcastTo($channel, $data)
{ {
if (empty(self::$channels[$channel])) return; if (!isset(self::$channels[$channel])) return;
$text = is_string($data) ? $data : json_encode($data); $json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
foreach (self::$channels[$channel] as $sk => $_) { foreach (self::$channels[$channel] as $sk => $_) {
if (!isset(self::$clients[$sk]) || !is_resource(self::$clients[$sk]['socket'])) { if (isset(self::$clients[$sk])) {
self::encodeAndSend(self::$clients[$sk]['socket'], 0x1, $json);
}
}
}
static function subscribe($sk, $channel)
{
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($sk, $channel)
{
unset(self::$channels[$channel][$sk]); unset(self::$channels[$channel][$sk]);
continue; if (empty(self::$channels[$channel])) unset(self::$channels[$channel]);
if (isset(self::$clients[$sk])) unset(self::$clients[$sk]['channels'][$channel]);
self::notifyRoomLeave($channel, $sk);
} }
self::encodeAndSend(self::$clients[$sk]['socket'], 0x1, $text);
}
}
// ── Channels ─────────────────────────────────────────
static function subscribe($socketKey, $channel)
{
self::$channels[$channel][$socketKey] = true;
self::$clients[$socketKey]['channels'][$channel] = true;
}
static function unsubscribe($socketKey, $channel)
{
unset(self::$channels[$channel][$socketKey]);
unset(self::$clients[$socketKey]['channels'][$channel]);
}
// ── Connection management ────────────────────────────
static function disconnect($sk) static function disconnect($sk)
{ {
if (!isset(self::$clients[$sk])) return; if (!isset(self::$clients[$sk])) return;
// Notify worker process if one exists for this socket
self::notifyDisconnect($sk); self::notifyDisconnect($sk);
$w = self::$clients[$sk]['watcher']; $w = self::$clients[$sk]['watcher'];
if ($w) Q_Evented::cancel($w); if ($w) Q_Evented::cancel($w);
foreach (self::$clients[$sk]['channels'] as $ch => $_) { foreach (self::$clients[$sk]['channels'] as $ch => $_) {
unset(self::$channels[$ch][$sk]); unset(self::$channels[$ch][$sk]);
self::notifyRoomLeave($ch, $sk);
} }
@fclose(self::$clients[$sk]['socket']); @fclose(self::$clients[$sk]['socket']);
unset(self::$clients[$sk]); 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) static function encodeAndSend($socket, $opcode, $payload)
{ {
$len = strlen($payload); $len = strlen($payload);
@@ -303,41 +192,37 @@ class Q_WebSocket
@fwrite($socket, $frame); @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 = '/') static function dispatchEvent($socketKey, $raw, $path = '/')
{ {
$msg = json_decode($raw, true); $msg = json_decode($raw, true);
if (!$msg || empty($msg['event'])) return; 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])) { if (!isset(self::$workers[$socketKey])) {
self::spawnWorker($socketKey, $path); self::spawnWorker($socketKey, $path);
} }
if (!isset(self::$workers[$socketKey])) return;
if (!isset(self::$workers[$socketKey])) return; // fork failed $json = json_encode($msg, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
// Forward message to child via length-prefixed JSON
$json = json_encode($msg, JSON_UNESCAPED_SLASHES);
$packet = pack('N', strlen($json)) . $json; $packet = pack('N', strlen($json)) . $json;
$written = @fwrite(self::$workers[$socketKey]['pipe'], $packet); $written = @fwrite(self::$workers[$socketKey]['pipe'], $packet);
if ($written === false || $written === 0) { if ($written === false || $written === 0) {
// Child died — respawn and retry once
self::cleanupWorker($socketKey); self::cleanupWorker($socketKey);
self::spawnWorker($socketKey, $path); self::spawnWorker($socketKey, $path);
if (isset(self::$workers[$socketKey])) { 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) static function spawnWorker($socketKey, $path)
{ {
if (!function_exists('pcntl_fork')) { if (!function_exists('pcntl_fork')) return;
return; // no fork — messages dispatch in-process
}
$pf = defined('STREAM_PF_UNIX') ? STREAM_PF_UNIX : STREAM_PF_INET; $pf = defined('STREAM_PF_UNIX') ? STREAM_PF_UNIX : STREAM_PF_INET;
$pair = stream_socket_pair($pf, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); $pair = stream_socket_pair($pf, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
if (!$pair) return; if (!$pair) return;
$pid = pcntl_fork(); $pid = pcntl_fork();
if ($pid === -1) { if ($pid === -1) { fclose($pair[0]); fclose($pair[1]); return; }
fclose($pair[0]); fclose($pair[1]);
return;
}
if ($pid === 0) { if ($pid === 0) {
// ── CHILD: message loop ── // ── CHILD: connection message loop ──
fclose($pair[0]); fclose($pair[0]);
$pipe = $pair[1]; $pipe = $pair[1];
Q_Socket::$_pipe = $pipe; Q_Socket::$_pipe = $pipe;
Q_Socket::$_socketId = $socketKey; Q_Socket::$_socketId = $socketKey;
// Fire _connect event
$connectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_connect', null); $connectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_connect', null);
if ($connectHandler) { if ($connectHandler) {
Q::event($connectHandler, array( Q::event($connectHandler, array(
@@ -388,30 +258,24 @@ class Q_WebSocket
Q_Socket::flush(); Q_Socket::flush();
} }
// Message loop — blocks on pipe reads, dispatches Q::event()
while (true) { while (true) {
$header = @fread($pipe, 4); $header = @fread($pipe, 4);
if ($header === false || $header === '' || strlen($header) < 4) break; if ($header === false || $header === '' || strlen($header) < 4) break;
$len = unpack('N', $header)[1]; $len = unpack('N', $header)[1];
if ($len <= 0 || $len > 10485760) break; if ($len <= 0 || $len > 10485760) break;
$json = ''; $json = '';
while (strlen($json) < $len) { while (strlen($json) < $len) {
$chunk = @fread($pipe, $len - strlen($json)); $chunk = @fread($pipe, $len - strlen($json));
if ($chunk === false || $chunk === '') break 2; if ($chunk === false || $chunk === '') break 2;
$json .= $chunk; $json .= $chunk;
} }
$msg = json_decode($json, true); $msg = json_decode($json, true);
if (!$msg) continue; if (!$msg) continue;
$event = $msg['event'] ?? ''; $event = $msg['event'] ?? '';
if ($event === '_disconnect') break; if ($event === '_disconnect') break;
// Resolve handler via config, or use event name directly
$mapped = Q_Config::get('Q', 'webserver', 'sockets', 'events', $event, $event); $mapped = Q_Config::get('Q', 'webserver', 'sockets', 'events', $event, $event);
Q_Socket::$_ack = isset($msg['ack']) ? $msg['ack'] : null; Q_Socket::$_ack = isset($msg['ack']) ? $msg['ack'] : null;
$result = null; $result = null;
@@ -422,18 +286,14 @@ class Q_WebSocket
'event' => $event, 'event' => $event,
'data' => $msg['data'] ?? array(), 'data' => $msg['data'] ?? array(),
); );
Q::event($mapped, $params, false, false, $result); Q::event($mapped, $params, false, false, $result);
// Auto-ack if handler returned a result
if (Q_Socket::$_ack !== null && $result !== null) { if (Q_Socket::$_ack !== null && $result !== null) {
Q_Socket::reply(array('ack' => Q_Socket::$_ack, 'data' => $result)); Q_Socket::reply(array('ack' => Q_Socket::$_ack, 'data' => $result));
} }
Q_Socket::flush(); Q_Socket::flush();
} }
// Fire _disconnect event
$disconnectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_disconnect', null); $disconnectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_disconnect', null);
if ($disconnectHandler) { if ($disconnectHandler) {
Q::event($disconnectHandler, array( Q::event($disconnectHandler, array(
@@ -441,7 +301,6 @@ class Q_WebSocket
)); ));
Q_Socket::flush(); Q_Socket::flush();
} }
fclose($pipe); fclose($pipe);
exit(0); exit(0);
} }
@@ -449,7 +308,6 @@ class Q_WebSocket
// ── PARENT ── // ── PARENT ──
fclose($pair[1]); fclose($pair[1]);
stream_set_blocking($pair[0], false); stream_set_blocking($pair[0], false);
$ipcWatcher = Q_Evented::onReadable($pair[0], function ($pipe) use ($socketKey) { $ipcWatcher = Q_Evented::onReadable($pair[0], function ($pipe) use ($socketKey) {
$data = @fread($pipe, 65536); $data = @fread($pipe, 65536);
if ($data === false || $data === '') { if ($data === false || $data === '') {
@@ -463,74 +321,321 @@ class Q_WebSocket
if ($cmd) Q_WebSocket::executeCommand($cmd); if ($cmd) Q_WebSocket::executeCommand($cmd);
} }
}); });
self::$workers[$socketKey] = array( self::$workers[$socketKey] = array(
'pid' => $pid, 'pid' => $pid, 'pipe' => $pair[0], 'watcher' => $ipcWatcher,
'pipe' => $pair[0],
'watcher' => $ipcWatcher,
); );
} }
/**
* Clean up a worker process for a socket.
* @method cleanupWorker
* @static
*/
static function cleanupWorker($socketKey) static function cleanupWorker($socketKey)
{ {
if (!isset(self::$workers[$socketKey])) return; if (!isset(self::$workers[$socketKey])) return;
$w = self::$workers[$socketKey]; $w = self::$workers[$socketKey];
if ($w['watcher']) Q_Evented::cancel($w['watcher']); if ($w['watcher']) Q_Evented::cancel($w['watcher']);
@fclose($w['pipe']); @fclose($w['pipe']);
if ($w['pid'] > 0) { if ($w['pid'] > 0 && function_exists('posix_kill')) {
if (function_exists("posix_kill")) posix_kill($w["pid"], SIGTERM); posix_kill($w['pid'], SIGTERM);
pcntl_waitpid($w['pid'], $st, WNOHANG); pcntl_waitpid($w['pid'], $st, WNOHANG);
} }
unset(self::$workers[$socketKey]); 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) static function notifyDisconnect($socketKey)
{ {
if (!isset(self::$workers[$socketKey])) return; 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())); $json = json_encode(array('event' => '_disconnect', 'data' => array()));
$packet = pack('N', strlen($json)) . $json; $packet = pack('N', strlen($json)) . $json;
@fwrite(self::$workers[$socketKey]['pipe'], $packet); @fwrite(self::$workers[$socketKey]['pipe'], $packet);
// Give child a moment then cleanup
self::cleanupWorker($socketKey); self::cleanupWorker($socketKey);
} }
// ── Room workers (process-per-room) ─────────────
/** /**
* Run a socket event handler in-process (Windows/no fork fallback). * Get room patterns from config. Cached.
* @method dispatchEventInProcess * Config format:
* Q.webserver.sockets.rooms.$pattern = {handler, tick?}
* e.g. "game/$id" => {"handler": "game/room", "tick": 100}
* @method getRoomPatterns
* @static * @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) static function dispatchEventInProcess($eventName, $params, $socketKey, $ack)
{ {
Q_Socket::$_directMode = true; Q_Socket::$_directMode = true;
Q_Socket::$_socketId = $socketKey; Q_Socket::$_socketId = $socketKey;
Q_Socket::$_ack = $ack; Q_Socket::$_ack = $ack;
$result = null; $result = null;
Q::event($eventName, $params, false, false, $result); Q::event($eventName, $params, false, false, $result);
if ($ack !== null && $result !== null) { if ($ack !== null && $result !== null) {
self::send($socketKey, array('ack' => $ack, 'data' => $result)); self::send($socketKey, array('ack' => $ack, 'data' => $result));
} }
Q_Socket::$_directMode = false; Q_Socket::$_directMode = false;
} }
/** // ── IPC command execution ───────────────────────
* Execute an IPC command from a child process.
* @method executeCommand
* @static
*/
static function executeCommand($cmd) static function executeCommand($cmd)
{ {
switch ($cmd['cmd'] ?? '') { switch ($cmd['cmd'] ?? '') {
+45 -24
View File
@@ -1,23 +1,32 @@
/** /**
* QSocket tiny WebSocket client for Qbix Server. * QSocket WebSocket client for Qbix Server.
* *
* Usage: * Usage:
* var qs = new QSocket('ws://localhost:8080/ws/chat'); * var qs = new QSocket('ws://localhost:8080/ws/chat');
* *
* // Listen for events from the server
* qs.on('chat/message', function(data) { * 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) { * // Send event with callback (server acks with structured data)
* console.log('Server confirmed:', ack); * 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): * Protocol (JSON over WebSocket):
* Client Server: {"event": "...", "data": {...}, "ack": N} * Client Server: {"event": "...", "data": {...}, "ack": N}
* Server Client: {"event": "...", "data": {...}} (broadcast) * Server Client: {"event": "...", "data": {...}} (broadcast)
* Server Client: {"ack": N, "data": {...}} (callback) * 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) { (function (root) {
'use strict'; 'use strict';
@@ -39,38 +48,31 @@
self.ws.onopen = function () { self.ws.onopen = function () {
self._delay = self._reconnectDelay; self._delay = self._reconnectDelay;
// Flush queued messages
while (self._queue.length) { while (self._queue.length) {
self.ws.send(self._queue.shift()); self.ws.send(self._queue.shift());
} }
if (self._handlers['connect']) { self._fire('connect');
self._handlers['connect'].forEach(function (fn) { fn(); });
}
}; };
self.ws.onmessage = function (e) { self.ws.onmessage = function (e) {
var msg; var msg;
try { msg = JSON.parse(e.data); } catch (err) { return; } 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]) { if (msg.ack !== undefined && self._acks[msg.ack]) {
self._acks[msg.ack](msg.data); self._acks[msg.ack](msg.data);
delete self._acks[msg.ack]; delete self._acks[msg.ack];
return; return;
} }
// Event broadcast // Event broadcast — pass full data to all listeners
if (msg.event && self._handlers[msg.event]) { if (msg.event) {
self._handlers[msg.event].forEach(function (fn) { self._fire(msg.event, msg.data);
fn(msg.data);
});
} }
}; };
self.ws.onclose = function () { self.ws.onclose = function () {
if (self._handlers['disconnect']) { self._fire('disconnect');
self._handlers['disconnect'].forEach(function (fn) { fn(); });
}
if (self._reconnect) { if (self._reconnect) {
self._delay = Math.min(self._delay * 1.5, self._maxDelay); self._delay = Math.min(self._delay * 1.5, self._maxDelay);
setTimeout(self._connect, self._delay); 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._delay = self._reconnectDelay;
self._connect(); self._connect();
} }
/** /**
* Listen for an event from the server. * 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) { QSocket.prototype.on = function (event, fn) {
if (!this._handlers[event]) this._handlers[event] = []; 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) { QSocket.prototype.off = function (event, fn) {
if (!this._handlers[event]) return this; if (!this._handlers[event]) return this;
if (!fn) { delete 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; return this;
}; };
/** /**
* Send an event to the server. * 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) { 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') { if (typeof callback === 'function') {
msg.ack = ++this._ackId; msg.ack = ++this._ackId;
this._acks[msg.ack] = callback; 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 () { QSocket.prototype.close = function () {
this._reconnect = false; this._reconnect = false;