diff --git a/README.md b/README.md index 1f40706..3a367f3 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ Same hardware. Same PHP code. **10x more users served.** | πŸ‘₯ **Concurrent PHP** (8GB) | ~160 workers | **~1,600 workers** | | πŸ”’ **Access-controlled files** | Public URLs or hacky rewrites | `X-Accel-Redirect` β€” PHP checks access, server streams the file | | 🧩 **Cache invalidation** | Whole-page only (purge everything) | `X-Cache-Tree` β€” invalidate one component, keep the rest cached | -| 🌐 **WebSocket** | Needs a separate server | Built in | +| 🌐 **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). @@ -50,6 +50,9 @@ dramatically faster and more scalable. - [vs FrankenPHP and Swoole](#️-vs-frankenphp-and-swoole) - [Features](#-features) - [Server Headers](#-server-headers--what-your-php-can-send) +- [WebSocket β€” Real-Time PHP](#-websocket--real-time-php) +- [Example: A Complete Chat App](#-example-a-complete-chat-app) +- [Clean URL Routing](#️-clean-url-routing-optional) - [For PHP Developers](#-for-php-developers--the-micro-framework) - [Configuration](#-configuration) - [Three Ways to Run](#-three-ways-to-run) @@ -83,7 +86,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, 196KB) +# Or use the PHAR (single file, ~250KB) php bin/qbixserver.phar --root=./public --port=8080 ``` @@ -463,6 +466,468 @@ update themselves without any manual invalidation calls. --- +## πŸ”Œ WebSocket β€” Real-Time PHP + +Each WebSocket connection gets **one PHP process** β€” forked from the preloaded +parent, stays alive for the entire connection. The server dispatches each message +to a handler via `Q::event()`. Static variables in handlers persist across +messages. When the client disconnects, the process exits β€” all state wiped. + +Same mental model as HTTP handlers, same `handlers/` directory, same `Q::event()`. +The only difference: the process lives longer. + +### Handlers + +```php + 'chat/message', + 'data' => ['user' => $userId, 'text' => $text], + ]); + + $result = ['count' => $messageCount]; +} +``` + +```php + $params['data']['room']]; +} +``` + +```php + 'invalid token']); + return; + } + Q_Socket::join($params['_socketId'], "user/$userId"); + $result = ['authenticated' => true]; +} +``` + +### Config + +Map event names to handlers. Also supports `_connect` and `_disconnect` lifecycle events: + +```json +{ + "Q": { + "webserver": { + "sockets": { + "events": { + "_connect": "auth/connect", + "_disconnect": "chat/leave", + "chat/message": "chat/message", + "chat/join": "chat/join", + "chat/typing": "chat/typing" + } + } + } + } +} +``` + +If no mapping is configured, the event name is used directly as the handler path. + +### The JS client (qbix-socket.js) + +```html + + +``` + +Auto-reconnects with exponential backoff. Ack callbacks for request-response. + +### Q_Socket API + +| Method | What it does | +|---|---| +| `Q_Socket::reply($data)` | Send to this connection's client | +| `Q_Socket::send($socketId, $data)` | Send to a specific client | +| `Q_Socket::broadcast($room, $data)` | Send to all clients in a room | +| `Q_Socket::broadcastAll($data)` | Send to ALL connected clients | +| `Q_Socket::join($socketId, $room)` | Subscribe a client to a room | +| `Q_Socket::leave($socketId, $room)` | Unsubscribe from a room | + +### Protocol + +``` +Client β†’ Server: {"event": "chat/message", "data": {...}, "ack": 42} +Server β†’ Client: {"ack": 42, "data": {...}} (callback) +Server β†’ Client: {"event": "chat/message", "data": {...}} (broadcast) +``` + +### Architecture + +``` +Browser ←─WebSocket─→ Parent (event loop) + β”‚ + connect: fork child, create IPC pipe + message: parent writes to child's pipe + child runs Q::event() handler + child calls Q_Socket::broadcast() + parent reads pipe, sends to sockets + disconnect: parent signals, child exits +``` + +One process per connection. Each handler is a thin wrapper calling preloaded +class methods β€” the per-connection COW delta is typically ~40-200KB (just +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. + +--- + +## πŸ“– Example: A Complete Chat App + +Everything below fits in one small project. HTTP handles pages and REST. +WebSocket handles real-time messaging. Both use the same `classes/` and +`handlers/` directories. + +### Project structure + +``` +chat/ +β”œβ”€β”€ qbixserver.php +β”œβ”€β”€ config/ +β”‚ └── server.json +β”œβ”€β”€ web/ +β”‚ β”œβ”€β”€ index.html ← static: the chat UI +β”‚ β”œβ”€β”€ qbix-socket.js ← static: WebSocket client +β”‚ β”œβ”€β”€ api/ +β”‚ β”‚ β”œβ”€β”€ messages.php ← HTTP: GET recent messages +β”‚ β”‚ └── login.php ← HTTP: POST authenticate, return token +β”‚ └── style.css +β”œβ”€β”€ classes/ +β”‚ └── Chat/ +β”‚ β”œβ”€β”€ Auth.php ← shared: token validation +β”‚ β”œβ”€β”€ Messages.php ← shared: DB read/write +β”‚ └── Rooms.php ← shared: room membership +└── handlers/ + └── chat/ + β”œβ”€β”€ connect.php ← socket: authenticate on connect + β”œβ”€β”€ disconnect.php ← socket: set user offline + β”œβ”€β”€ message.php ← socket: broadcast a message + β”œβ”€β”€ join.php ← socket: join a room + └── typing.php ← socket: broadcast typing indicator +``` + +### Config + +```json +{ + "Q": { + "webserver": { + "preload": { + "classes": ["Chat\\Auth", "Chat\\Messages", "Chat\\Rooms"] + }, + "sockets": { + "events": { + "_connect": "chat/connect", + "_disconnect": "chat/disconnect", + "chat/message":"chat/message", + "chat/join": "chat/join", + "chat/typing": "chat/typing" + } + } + } + } +} +``` + +### HTTP scripts β€” pages and REST + +```php + 'Invalid credentials']); + exit; +} + +header('Content-Type: application/json'); +echo json_encode([ + 'token' => Chat\Auth::createToken($user['id']), + 'userId' => $user['id'], + 'name' => $user['name'], +]); +``` + +```php + 'not authenticated']); + return; + } + $userId = $user['id']; + $userName = $user['name']; + } + + $text = $params['data']['text'] ?? ''; + if (!$text) return; + + // Save to database + $id = Chat\Messages::save($userId, $params['data']['room'] ?? 'general', $text); + + // Broadcast to everyone in the room + Q_Socket::broadcast($params['data']['room'] ?? 'general', [ + 'event' => 'chat/message', + 'data' => [ + 'id' => $id, + 'user' => $userName, + 'text' => $text, + 'time' => date('c'), + ], + ]); + + $result = ['id' => $id]; // ack back to sender +} +``` + +```php + $room]; +} +``` + +```php + 'chat/typing', + 'data' => ['user' => $params['data']['user']], + ]); +} +``` + +```php + 'Authentication required']); + exit; // safe β€” forked process + } +} +``` + +```php += 2 + && (($value[0] === '"' && $value[strlen($value)-1] === '"') + || ($value[0] === "'" && $value[strlen($value)-1] === "'")) + ) { + $value = substr($value, 1, -1); + } + $_ENV[$name] = $value; + $_SERVER[$name] = $value; + putenv("$name=$value"); + } +} + // ── Load config ───────────────────────────────────── // Default server config diff --git a/src/Q.php b/src/Q.php index b5442d6..6d6a2e8 100644 --- a/src/Q.php +++ b/src/Q.php @@ -398,6 +398,274 @@ class Q spl_autoload_register(array('Q', 'autoload')); +// ── Q_Socket ──────────────────────────────────────── + +/** + * PHP API for WebSocket handlers β€” sending messages, managing rooms. + * + * Each WebSocket connection gets one PHP process. The server dispatches + * messages via Q::event() to handlers. Handlers use Q_Socket to send + * data back. Static variables in handlers persist across messages + * (same process) and are wiped on disconnect (process dies). + * + * @class Q_Socket + */ +class Q_Socket +{ + /** @var resource IPC pipe to parent */ + static $_pipe = null; + /** @var integer Current client's socket key */ + static $_socketId = null; + /** @var integer|null Ack ID from current message */ + static $_ack = null; + /** @var boolean True when running in-process (no fork) */ + static $_directMode = false; + /** @var array Buffered outbound commands */ + static $_buffer = array(); + + /** + * Send data to the client that owns this connection. + */ + static function reply($data) + { + self::send(self::$_socketId, $data); + } + + /** + * Send data to a specific connected client. + */ + static function send($socketId, $data) + { + self::_command(array('cmd' => 'send', 'socketId' => $socketId, 'data' => $data)); + } + + /** + * Broadcast to all clients in a room/channel. + */ + static function broadcast($room, $data) + { + self::_command(array('cmd' => 'broadcast', 'room' => $room, 'data' => $data)); + } + + /** + * Broadcast to ALL connected WebSocket clients. + */ + static function broadcastAll($data) + { + self::_command(array('cmd' => 'broadcastAll', 'data' => $data)); + } + + /** + * Subscribe a client to a room/channel. + */ + static function join($socketId, $room) + { + self::_command(array('cmd' => 'join', 'socketId' => $socketId, 'room' => $room)); + } + + /** + * Unsubscribe a client from a room/channel. + */ + static function leave($socketId, $room) + { + self::_command(array('cmd' => 'leave', 'socketId' => $socketId, 'room' => $room)); + } + + /** + * Buffer a command or execute directly in-process. + */ + private static function _command($cmd) + { + if (self::$_directMode) { + Q_WebSocket::executeCommand($cmd); + } else { + self::$_buffer[] = $cmd; + } + } + + /** + * Flush buffered commands to the IPC pipe. + * Called automatically after each handler invocation. + */ + static function flush() + { + if (!self::$_pipe || empty(self::$_buffer)) return; + $out = ''; + foreach (self::$_buffer as $cmd) { + $out .= json_encode($cmd, JSON_UNESCAPED_SLASHES) . "\n"; + } + @fwrite(self::$_pipe, $out); + self::$_buffer = array(); + } +} + +// ── Q_Request ─────────────────────────────────────── + +/** + * Minimal Q_Request β€” compatible subset of the Qbix Platform's Q_Request. + * Provides convenient access to request data that the server has already parsed. + * + * @class Q_Request + */ +class Q_Request +{ + /** + * Raw request body. Set by the server before your script runs. + * Use this instead of php://input (which doesn't work in our model). + * @property $input + * @type string + * @static + */ + static $input = ''; + + /** + * Get the HTTP method (GET, POST, PUT, DELETE, etc.) + * @method method + * @static + * @return {string} + */ + static function method() + { + return $_SERVER['REQUEST_METHOD'] ?? 'GET'; + } + + /** + * Get the raw request body. + * @method input + * @static + * @return {string} + */ + static function input() + { + return self::$input; + } + + /** + * Get the request body parsed as JSON. + * @method json + * @static + * @param {boolean} $assoc Return associative array (default true) + * @return {array|object|null} + */ + static function json($assoc = true) + { + return json_decode(self::$input, $assoc); + } + + /** + * Get the full request URL. + * @method url + * @static + * @param {boolean} $querystring Include query string (default true) + * @return {string} + */ + static function url($querystring = true) + { + $scheme = ($_SERVER['REQUEST_SCHEME'] ?? 'http'); + $host = $_SERVER['HTTP_HOST'] ?? $_SERVER['SERVER_NAME'] ?? 'localhost'; + $uri = $querystring + ? ($_SERVER['REQUEST_URI'] ?? '/') + : ($_SERVER['SCRIPT_NAME'] ?? '/'); + return $scheme . '://' . $host . $uri; + } + + /** + * Get the URL path (without query string). + * @method path + * @static + * @return {string} + */ + static function path() + { + $uri = $_SERVER['REQUEST_URI'] ?? '/'; + $qPos = strpos($uri, '?'); + return $qPos !== false ? substr($uri, 0, $qPos) : $uri; + } + + /** + * Get a request header value. + * @method header + * @static + * @param {string} $name Header name (case-insensitive) + * @return {string|null} + */ + static function header($name) + { + $key = 'HTTP_' . strtoupper(str_replace('-', '_', $name)); + return $_SERVER[$key] ?? null; + } + + /** + * Get the client's IP address (resolved through proxy headers by the server). + * @method ip + * @static + * @return {string} + */ + static function ip() + { + return $_SERVER['REMOTE_ADDR'] ?? '0.0.0.0'; + } + + /** + * Check if the request is an AJAX/XHR request. + * @method isAjax + * @static + * @return {boolean} + */ + static function isAjax() + { + return strtolower($_SERVER['HTTP_X_REQUESTED_WITH'] ?? '') === 'xmlhttprequest'; + } + + /** + * Get uploaded files. Convenience wrapper around $_FILES. + * @method files + * @static + * @param {string|null} $name Specific file input name, or null for all + * @return {array|null} + */ + static function files($name = null) + { + if ($name === null) return $_FILES; + return $_FILES[$name] ?? null; + } + + /** + * Get the Content-Type of the request. + * @method contentType + * @static + * @return {string} + */ + static function contentType() + { + return $_SERVER['CONTENT_TYPE'] ?? $_SERVER['HTTP_CONTENT_TYPE'] ?? ''; + } + + /** + * Check if the request body is JSON. + * @method isJson + * @static + * @return {boolean} + */ + static function isJson() + { + return strpos(strtolower(self::contentType()), 'application/json') !== false; + } + + /** + * Get a value from $_GET, $_POST, or $_REQUEST with a default. + * @method special + * @static + * @param {string} $name + * @param {mixed} $default + * @return {mixed} + */ + static function special($name, $default = null) + { + return $_REQUEST[$name] ?? $default; + } +} + // ── Q_Config ──────────────────────────────────────── /** diff --git a/src/Q/Uri.php b/src/Q/Uri.php new file mode 100644 index 0000000..ec374ac --- /dev/null +++ b/src/Q/Uri.php @@ -0,0 +1,311 @@ +fields[$name] ?? null; + } + + function __set($name, $value) + { + $this->fields[$name] = $value; + } + + function __isset($name) + { + return isset($this->fields[$name]); + } + + function toArray() + { + return $this->fields; + } + + /** + * Create a Q_Uri from an array of fields. + * @method from + * @static + */ + static function from($fields) + { + $uri = new self(); + if (is_array($fields)) { + $uri->fields = $fields; + } + return $uri; + } + + /** + * Get merged routes from Q/routes@start, Q/routes, Q/routes@end. + * Same merge order as the full Platform. Memoized. + * @method getRoutes + * @static + * @return {array} + */ + static function getRoutes() + { + if (isset(self::$routesCache)) { + return self::$routesCache; + } + $routesStart = Q_Config::get('Q', 'routes@start', array()); + $routes = Q_Config::get('Q', 'routes', array()); + $routesEnd = Q_Config::get('Q', 'routes@end', array()); + // Reverse order within each block (later plugins override earlier) + $result = array(); + foreach (array($routesStart, $routes, $routesEnd) as $source) { + if (!is_array($source)) continue; + $keys = array_keys($source); + $vals = array_values($source); + $keys = array_reverse($keys); + $vals = array_reverse($vals); + foreach ($keys as $i => $k) { + if (!isset($result[$k])) { + $result[$k] = $vals[$i]; + } + } + } + self::$routesCache = $result; + return $result; + } + + /** + * Clear all memoized routing state. + * Call when config changes (e.g. --hot reload). + * @method clearRouteCache + * @static + */ + static function clearRouteCache() + { + self::$routesCache = null; + self::$compiledPatterns = array(); + self::$pathCache = array(); + } + + /** + * Route a URL path to a Q_Uri using configured routes. + * Results are memoized β€” the same path always returns the same URI. + * @method fromPath + * @static + * @param {string} $path URL path (e.g. "api/users/42") + * @return {Q_Uri|null} + */ + static function fromPath($path) + { + $path = trim($path, '/'); + if (isset(self::$pathCache[$path])) { + return self::$pathCache[$path]; + } + + $segments = $path !== '' ? explode('/', $path) : array(); + $routes = self::getRoutes(); + if (empty($routes)) { + self::$pathCache[$path] = null; + return null; + } + + foreach ($routes as $pattern => $fields) { + if (!isset($fields)) continue; // disabled route + + $matched = self::matchSegments($pattern, $segments); + if ($matched === false) continue; + + // Check regex constraints on matched values + $valid = true; + foreach ($matched as $k => $v) { + if (isset($fields[$k]) && is_string($fields[$k])) { + if (!preg_match('/' . $fields[$k] . '/', $v)) { + $valid = false; + break; + } + } + } + // Special condition handler (same as Platform) + if ($valid && !empty($fields[''])) { + $params = array( + 'uriFields' => $matched, + 'routeFields' => $fields, + 'fields' => array_merge($fields, $matched), + 'pattern' => $pattern, + ); + if (false === Q::event($fields[''], $params, false, false, $params)) { + $valid = false; + } + } + if (!$valid) continue; + + // Merge route defaults with matched values + $uriFields = array(); + foreach ($fields as $k => $v) { + if ($k === '' || is_int($k)) continue; + $uriFields[$k] = $v; + } + $uriFields = array_merge($uriFields, $matched); + + $uri = new self(); + $uri->fields = $uriFields; + $uri->route = $pattern; + self::$pathCache[$path] = $uri; + return $uri; + } + + self::$pathCache[$path] = null; + return null; + } + + /** + * Compile a route pattern into a reusable structure. + * Same implementation as the full Platform's Q_Uri::compilePattern(). + * Memoized by pattern string β€” compiled once, reused forever. + * @method compilePattern + * @static + * @protected + */ + protected static function compilePattern($pattern) + { + if (isset(self::$compiledPatterns[$pattern])) { + return self::$compiledPatterns[$pattern]; + } + $route_segments = explode('/', $pattern); + $tailArray = false; + $tailField = null; + $valid = true; + if (substr($pattern, -2) === '[]') { + $tailArray = true; + $last_rs = end($route_segments); + if (!isset($last_rs[0]) || !in_array($last_rs[0], self::$variablePrefixes)) { + $valid = false; + } else { + $tailField = substr($last_rs, 1, -2); + } + $route_segments = array_slice($route_segments, 0, -1); + } + $segments = array(); + foreach ($route_segments as $rs) { + $rs_parts = explode('.', $rs); + $parts = array(); + foreach ($rs_parts as $part) { + if (!isset($part[0]) || !in_array($part[0], self::$variablePrefixes)) { + $parts[] = array( + 'var' => false, + 'literal' => str_replace( + self::$escapedVariablePrefixes, + self::$variablePrefixes, + $part + ) + ); + } else { + $parts[] = array( + 'var' => true, + 'field' => substr($part, 1) + ); + } + } + $segments[] = $parts; + } + $compiled = array( + 'valid' => $valid, + 'segments' => $segments, + 'count' => count($segments), + 'tailArray' => $tailArray, + 'tailField' => $tailField, + ); + self::$compiledPatterns[$pattern] = $compiled; + return $compiled; + } + + /** + * Match URL segments against a compiled route pattern. + * Same implementation as the full Platform's Q_Uri::matchSegments(). + * @method matchSegments + * @static + * @protected + */ + protected static function matchSegments($pattern, $segments) + { + if (!$pattern && $pattern !== '0') { + return count($segments) === 0 ? array() : false; + } + $compiled = self::compilePattern($pattern); + if (!$compiled['valid']) return false; + + $count = $compiled['count']; + $segCount = count($segments); + + if ($compiled['tailArray']) { + if ($count >= $segCount) return false; + } else { + if ($count !== $segCount) return false; + } + + $args = array(); + $cs = $compiled['segments']; + for ($i = 0; $i < $count; $i++) { + $rs_parts = $cs[$i]; + $rs_parts_count = count($rs_parts); + $segment = urldecode($segments[$i]); + $s_parts = explode('.', $segment, $rs_parts_count); + if (count($s_parts) < $rs_parts_count) return false; + + for ($j = 0; $j < $rs_parts_count; $j++) { + $p = $rs_parts[$j]; + if (!$p['var']) { + if ($s_parts[$j] !== $p['literal']) return false; + continue; + } + $args[$p['field']] = $s_parts[$j]; + } + } + + if ($compiled['tailArray']) { + $args[$compiled['tailField']] = array(); + for (; $i < $segCount; $i++) { + $args[$compiled['tailField']][] = urldecode($segments[$i]); + } + } + + return $args; + } +} diff --git a/src/Q/WebServer.php b/src/Q/WebServer.php index 1bb0c91..9696369 100644 --- a/src/Q/WebServer.php +++ b/src/Q/WebServer.php @@ -494,6 +494,9 @@ class Q_WebServer // Resolve proxy headers for real client IP $directIp = self::$clientInfo[$key]['ip'] ?? '0.0.0.0'; $parsed['clientIp'] = Q_WebServer_Proxy::clientIp($directIp, $parsed['headers']); + $parsed['_remoteAddr'] = $parsed['clientIp']; + $peer = stream_socket_get_name($client, true); + $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); @@ -765,7 +768,19 @@ class Q_WebServer if ($handled) return false; } - // 2. Blocked paths + // 2. WebSocket upgrade on any path + $upgrade = strtolower($parsed['headers']['upgrade'] ?? ''); + if ($upgrade === 'websocket' && $path !== '/Q/ws') { + $upgraded = Q_WebSocket::upgrade( + $client, $parsed['headers'], + function ($sk, $msg) use ($path) { + Q_WebSocket::dispatchEvent($sk, $msg, $path); + } + ); + return $upgraded; + } + + // 3. Blocked paths if (self::isBlocked($path)) { self::sendResponse($client, 403, 'Forbidden'); return false; @@ -838,17 +853,190 @@ class Q_WebServer } } - // 6. Clean URL β†’ route through index.php + // 6. Route dispatch β€” if Q.routes configured, match URL to handler + // Q_Uri caches compiled patterns and pathβ†’URI results in memory. + static $routingEnabled = null; + if ($routingEnabled === null) { + $routingEnabled = Q_Config::get('Q', 'routes', null) !== null + && class_exists('Q_Uri', true); + } + if ($routingEnabled) { + $uri = Q_Uri::fromPath($path); + if ($uri && !empty($uri->module) && !empty($uri->action)) { + return self::handleRoute($client, $parsed, $uri); + } + } + + // 7. Clean URL β†’ route through index.php (if exists) $indexPhp = self::$rootDir . 'index.php'; if (is_file($indexPhp)) { return self::handlePhp($client, $parsed, $indexPhp); } - // 7. Not found + // 8. Not found self::sendResponse($client, 404, self::render404($path), 'text/html; charset=utf-8'); return false; } + /** + * Handle a routed request via Q::event() dispatch pipeline. + * Fires the same events as Qbix Platform's Q_Dispatcher: + * {module}/{action}/validate β†’ validate input + * {module}/{action}/{method} β†’ handle GET/POST/PUT/DELETE + * {module}/{action}/response β†’ render response + * + * @method handleRoute + * @static + * @private + * @param {resource} $client + * @param {array} $parsed + * @param {Q_Uri} $uri + * @return {boolean} + */ + private static function handleRoute($client, $parsed, $uri) + { + $module = $uri->module; + $action = $uri->action; + $routed = $uri->toArray(); + $method = strtolower($parsed['method']); // get, post, put, delete + + // Set up superglobals + $parsed['_scriptPath'] = ''; // no script β€” handler-based + $saved = array($_SERVER, $_GET, $_POST, $_REQUEST, $_COOKIE); $_SERVER['REQUEST_METHOD'] = $parsed['method']; + $_SERVER['REQUEST_URI'] = $parsed['uri']; + $_SERVER['QUERY_STRING'] = $parsed['query']; + $_SERVER['SERVER_NAME'] = explode(':', $parsed['headers']['host'] ?? 'localhost')[0]; + $_SERVER['SERVER_PORT'] = self::$port; + $_SERVER['SERVER_PROTOCOL'] = 'HTTP/1.1'; + $_SERVER['SERVER_SOFTWARE'] = 'QbixServer/1.0'; + $_SERVER['DOCUMENT_ROOT'] = rtrim(self::$rootDir, DS); + $_SERVER['REMOTE_ADDR'] = $parsed['_remoteAddr'] ?? '127.0.0.1'; + $_SERVER['REQUEST_TIME'] = time(); + $_SERVER['REQUEST_TIME_FLOAT'] = microtime(true); + foreach ($parsed['headers'] as $k => $v) { + $_SERVER['HTTP_' . strtoupper(str_replace('-', '_', $k))] = $v; + } + + $_GET = $_POST = $_REQUEST = $_FILES = array(); + if ($parsed['query']) parse_str($parsed['query'], $_GET); + $ct = strtolower($parsed['headers']['content-type'] ?? ''); + $rawBody = $parsed['body'] ?? ''; + if (strpos($ct, 'application/x-www-form-urlencoded') !== false) { + parse_str($rawBody, $_POST); + } elseif (strpos($ct, 'application/json') !== false) { + $_POST = json_decode($rawBody, true) ?: array(); + } elseif (strpos($ct, 'multipart/form-data') !== false) { + $origCt = $parsed['headers']['content-type'] ?? $_SERVER['CONTENT_TYPE'] ?? ''; + self::parseMultipart($origCt, $rawBody, $_POST, $_FILES); + } + $_REQUEST = array_merge($_GET, $_POST); + + // Make raw body available + Q_Request::$input = $rawBody; + + // If pcntl available, fork to isolate + if (function_exists('pcntl_fork')) { + $pid = pcntl_fork(); + if ($pid === 0) { + // ── CHILD: run dispatch pipeline ── + while (ob_get_level()) ob_end_clean(); + ob_start(); + $status = 200; + $headers = array(); + + try { + // 1. Validate + Q::event("$module/$action/validate", $routed, false, true); + + // 2. Method handler (get, post, put, delete) + if (Q::canHandle("$module/$action/$method")) { + Q::event("$module/$action/$method", $routed); + } elseif ($method !== 'get') { + $status = 405; + echo 'Method Not Allowed'; + } + + // 3. Response + Q::event("$module/$action/response", $routed, false, true); + + foreach (headers_list() as $h) { + if (strpos($h, ':') !== false) { + list($k, $v) = explode(':', $h, 2); + $headers[trim($k)] = trim($v); + } + } + $code = http_response_code(); + if ($code) $status = $code; + } catch (\Throwable $e) { + $status = 500; + ob_clean(); + echo json_encode(array('error' => $e->getMessage())); + $headers['Content-Type'] = 'application/json'; + } + + $body = ob_get_clean(); + $response = compact('status', 'body', 'headers'); + Q_WebServer_Headers::processResponse($client, $response, $parsed['headers']); + @fclose($client); + exit(0); + } elseif ($pid > 0) { + @fclose($client); + $key = (int) $client; + if (isset(self::$clientWatchers[$key])) { + Q_Evented::cancel(self::$clientWatchers[$key]); + } + unset(self::$clientWatchers[$key], self::$clients[$key], self::$buffers[$key]); + pcntl_waitpid($pid, $st, WNOHANG); + self::$lastStatus = 200; + list($_SERVER, $_GET, $_POST, $_REQUEST, $_COOKIE) = $saved; + return false; + } + // Fork failed β€” fall through to in-process + } + + // In-process fallback + while (ob_get_level()) ob_end_clean(); + header_remove(); + http_response_code(200); + ob_start(); + $status = 200; + $headers = array(); + + try { + Q::event("$module/$action/validate", $routed, false, true); + if (Q::canHandle("$module/$action/$method")) { + Q::event("$module/$action/$method", $routed); + } elseif ($method !== 'get') { + $status = 405; + echo 'Method Not Allowed'; + } + Q::event("$module/$action/response", $routed, false, true); + foreach (headers_list() as $h) { + if (strpos($h, ':') !== false) { + list($k, $v) = explode(':', $h, 2); + $headers[trim($k)] = trim($v); + } + } + $code = http_response_code(); + if ($code) $status = $code; + } catch (\Throwable $e) { + $status = 500; + ob_clean(); + echo json_encode(array('error' => $e->getMessage())); + $headers['Content-Type'] = 'application/json'; + } + + $body = ob_get_clean(); + header_remove(); + list($_SERVER, $_GET, $_POST, $_REQUEST, $_COOKIE) = $saved; + + $response = compact('status', 'body', 'headers'); + Q_WebServer_Headers::processResponse($client, $response, $parsed['headers']); + self::$lastStatus = $status; + Q_WebServer_Cache::put($parsed, $response); + return false; + } + /** * Route a .php script to the worker pool or dispatch in-process. * @return {boolean} false (connection closes after response) @@ -934,19 +1122,42 @@ $_SERVER['REQUEST_URI'] = $req['uri'] ?? '/'; $_SERVER['QUERY_STRING'] = $req['query'] ?? ''; $_SERVER['SCRIPT_FILENAME'] = $req['scriptPath'] ?? ''; $_SERVER['SCRIPT_NAME'] = '/' . basename($req['scriptPath'] ?? 'index.php'); +$_SERVER['PHP_SELF'] = $_SERVER['SCRIPT_NAME']; +$_SERVER['PATH_TRANSLATED'] = $req['scriptPath'] ?? ''; $_SERVER['DOCUMENT_ROOT'] = $req['documentRoot'] ?? ''; +$_SERVER['DOCUMENT_URI'] = $_SERVER['SCRIPT_NAME']; $_SERVER['SERVER_NAME'] = $req['serverName'] ?? 'localhost'; $_SERVER['SERVER_PORT'] = $req['serverPort'] ?? '8080'; +$_SERVER['SERVER_ADDR'] = '127.0.0.1'; +$_SERVER['SERVER_PROTOCOL'] = 'HTTP/1.1'; +$_SERVER['SERVER_SOFTWARE'] = 'QbixServer/1.0'; +$_SERVER['GATEWAY_INTERFACE'] = 'CGI/1.1'; +$_SERVER['REDIRECT_STATUS'] = 200; +$_SERVER['REMOTE_ADDR'] = $req['remoteAddr'] ?? '127.0.0.1'; +$_SERVER['REMOTE_PORT'] = $req['remotePort'] ?? 0; +$_SERVER['REQUEST_TIME'] = time(); +$_SERVER['REQUEST_TIME_FLOAT'] = microtime(true); +$_SERVER['REQUEST_SCHEME'] = ($req['https'] ?? false) ? 'https' : 'http'; +$_SERVER['HTTPS'] = ($req['https'] ?? false) ? 'on' : ''; foreach ($req['headers'] ?? [] as $k=>$v) $_SERVER['HTTP_'.strtoupper(str_replace('-','_',$k))] = $v; if (isset($req['headers']['content-type'])) $_SERVER['CONTENT_TYPE'] = $req['headers']['content-type']; if (isset($req['headers']['content-length'])) $_SERVER['CONTENT_LENGTH'] = $req['headers']['content-length']; -$_GET = $_POST = $_REQUEST = []; +// Parse cookies +$_COOKIE = []; +$ck = $req['headers']['cookie'] ?? ''; +if ($ck) { foreach (explode(';',$ck) as $p) { $p=trim($p); if(!$p)continue; $e=strpos($p,'='); if($e===false)continue; $_COOKIE[urldecode(trim(substr($p,0,$e)))]=urldecode(trim(substr($p,$e+1))); } } +// Parse Basic auth +$auth = $req['headers']['authorization'] ?? ''; +if (stripos($auth,'Basic ')===0) { $d=base64_decode(substr($auth,6)); if($d&&strpos($d,':')!==false) { [$u,$pw]=explode(':',$d,2); $_SERVER['PHP_AUTH_USER']=$u; $_SERVER['PHP_AUTH_PW']=$pw; $_SERVER['AUTH_TYPE']='Basic'; } } +$_GET = $_POST = $_REQUEST = $_FILES = []; if (!empty($req['query'])) parse_str($req['query'], $_GET); $ct = strtolower($_SERVER['CONTENT_TYPE'] ?? ''); $raw = $req['body'] ?? ''; if (strpos($ct,'application/x-www-form-urlencoded') !== false) parse_str($raw, $_POST); elseif (strpos($ct,'application/json') !== false) $_POST = json_decode($raw, true) ?: []; -$_REQUEST = array_merge($_GET, $_POST); +elseif (strpos($ct,'multipart/form-data') !== false) { $oct=$req['headers']['content-type']??''; Q_WebServer::parseMultipart($oct, $raw, $_POST, $_FILES); } +$_REQUEST = array_merge($_COOKIE, $_GET, $_POST); +if (class_exists('Q_Request',false)) Q_Request::$input = $raw; ob_start(); $status = 200; $headers = []; try { if (is_file($req['scriptPath'])) include $req['scriptPath']; else { $status = 404; echo 'Not Found'; } @@ -979,6 +1190,9 @@ WORKER; 'documentRoot'=> rtrim(self::$rootDir, DS), 'serverName' => explode(':', $host)[0], 'serverPort' => (string) self::$port, + 'remoteAddr' => $parsed['_remoteAddr'] ?? '127.0.0.1', + 'remotePort' => $parsed['_remotePort'] ?? 0, + 'https' => !empty(self::$tlsSocket), 'qFile' => $qFile, 'projectRoot' => dirname(rtrim(self::$rootDir, DS)), ), JSON_UNESCAPED_SLASHES); @@ -1408,33 +1622,106 @@ HTML static function dispatchToQ($parsed) { - $saved = array($_SERVER, $_GET, $_POST, $_REQUEST); - $_SERVER['REQUEST_METHOD'] = $parsed['method']; - $_SERVER['REQUEST_URI'] = $parsed['uri']; - $_SERVER['QUERY_STRING'] = $parsed['query']; - $_SERVER['SCRIPT_NAME'] = '/' . basename($parsed['_scriptPath'] ?? 'index.php'); - $_SERVER['SCRIPT_FILENAME'] = $parsed['_scriptPath'] ?? self::$rootDir . 'index.php'; + $saved = array($_SERVER, $_GET, $_POST, $_REQUEST, $_COOKIE); + $scriptPath = $parsed['_scriptPath'] ?? self::$rootDir . 'index.php'; $host = $parsed['headers']['host'] ?? 'localhost'; - $_SERVER['SERVER_NAME'] = explode(':', $host)[0]; // strip port from Host header - $_SERVER['SERVER_PORT'] = self::$port; - $_SERVER['DOCUMENT_ROOT'] = rtrim(self::$rootDir, DS); + $hostParts = explode(':', $host); + + // ── Standard CGI variables ────────────────────── + $_SERVER['REQUEST_METHOD'] = $parsed['method']; + $_SERVER['REQUEST_URI'] = $parsed['uri']; + $_SERVER['QUERY_STRING'] = $parsed['query']; + $_SERVER['SCRIPT_NAME'] = '/' . basename($scriptPath); + $_SERVER['SCRIPT_FILENAME'] = $scriptPath; + $_SERVER['PHP_SELF'] = $_SERVER['SCRIPT_NAME']; // WordPress uses this + $_SERVER['PATH_TRANSLATED'] = $scriptPath; + $_SERVER['PATH_INFO'] = ''; + $_SERVER['DOCUMENT_ROOT'] = rtrim(self::$rootDir, DS); + $_SERVER['DOCUMENT_URI'] = $_SERVER['SCRIPT_NAME']; + $_SERVER['SERVER_NAME'] = $hostParts[0]; + $_SERVER['SERVER_PORT'] = isset($hostParts[1]) ? $hostParts[1] : self::$port; + $_SERVER['SERVER_ADDR'] = self::$host === '0.0.0.0' ? '127.0.0.1' : self::$host; + $_SERVER['SERVER_PROTOCOL'] = 'HTTP/' . ($parsed['httpVersion'] ?? '1.1'); + $_SERVER['SERVER_SOFTWARE'] = 'QbixServer/' . (defined('QBIX_SERVER_VERSION') ? QBIX_SERVER_VERSION : '1.0'); + $_SERVER['GATEWAY_INTERFACE'] = 'CGI/1.1'; + $_SERVER['REDIRECT_STATUS'] = 200; + $_SERVER['REMOTE_ADDR'] = $parsed['_remoteAddr'] ?? '127.0.0.1'; + $_SERVER['REMOTE_PORT'] = $parsed['_remotePort'] ?? 0; + $_SERVER['REQUEST_TIME'] = time(); + $_SERVER['REQUEST_TIME_FLOAT']= microtime(true); + + // ── HTTPS detection (direct TLS or proxy header) ── + $isHttps = !empty(self::$tlsSocket); + $fwdProto = $parsed['headers']['x-forwarded-proto'] ?? ''; + if (strtolower($fwdProto) === 'https') $isHttps = true; + // CloudFront + $cfProto = $parsed['headers']['cloudfront-forwarded-proto'] ?? ''; + if (strtolower($cfProto) === 'https') $isHttps = true; + // Cloudflare + $cfVisitor = $parsed['headers']['cf-visitor'] ?? ''; + if (strpos($cfVisitor, '"https"') !== false) $isHttps = true; + $_SERVER['REQUEST_SCHEME'] = $isHttps ? 'https' : 'http'; + $_SERVER['HTTPS'] = $isHttps ? 'on' : ''; + + // ── Request headers β†’ HTTP_* ──────────────────── + // All request headers become HTTP_HEADERNAME (uppercase, hyphensβ†’underscores) foreach ($parsed['headers'] as $k => $v) { $_SERVER['HTTP_' . strtoupper(str_replace('-', '_', $k))] = $v; } + // Content-Type and Content-Length are special (no HTTP_ prefix per CGI spec) if (isset($parsed['headers']['content-type'])) $_SERVER['CONTENT_TYPE'] = $parsed['headers']['content-type']; if (isset($parsed['headers']['content-length'])) $_SERVER['CONTENT_LENGTH'] = $parsed['headers']['content-length']; - $_GET = $_POST = $_REQUEST = array(); + // ── Basic auth parsing ────────────────────────── + $auth = $parsed['headers']['authorization'] ?? ''; + if (stripos($auth, 'Basic ') === 0) { + $decoded = base64_decode(substr($auth, 6)); + if ($decoded && strpos($decoded, ':') !== false) { + list($user, $pass) = explode(':', $decoded, 2); + $_SERVER['PHP_AUTH_USER'] = $user; + $_SERVER['PHP_AUTH_PW'] = $pass; + $_SERVER['AUTH_TYPE'] = 'Basic'; + } + } elseif (stripos($auth, 'Bearer ') === 0) { + $_SERVER['HTTP_AUTHORIZATION'] = $auth; // already set by loop + $_SERVER['AUTH_TYPE'] = 'Bearer'; + } + + // ── $_COOKIE ──────────────────────────────────── + $_COOKIE = array(); + $cookieHeader = $parsed['headers']['cookie'] ?? ''; + if ($cookieHeader) { + $pairs = explode(';', $cookieHeader); + foreach ($pairs as $pair) { + $pair = trim($pair); + if ($pair === '') continue; + $eqPos = strpos($pair, '='); + if ($eqPos === false) continue; + $name = urldecode(trim(substr($pair, 0, $eqPos))); + $value = urldecode(trim(substr($pair, $eqPos + 1))); + $_COOKIE[$name] = $value; + } + } + + // ── $_GET, $_POST, $_FILES, $_REQUEST ─────────── + $_GET = $_POST = $_REQUEST = $_FILES = array(); if ($parsed['query']) parse_str($parsed['query'], $_GET); $ct = strtolower($_SERVER['CONTENT_TYPE'] ?? ''); + $rawBody = $parsed['body'] ?? ''; if (strpos($ct, 'application/x-www-form-urlencoded') !== false) { - parse_str($parsed['body'], $_POST); + parse_str($rawBody, $_POST); } elseif (strpos($ct, 'application/json') !== false) { - $_POST = json_decode($parsed['body'], true) ?: array(); + $_POST = json_decode($rawBody, true) ?: array(); + } elseif (strpos($ct, 'multipart/form-data') !== false) { + $origCt = $parsed['headers']['content-type'] ?? $_SERVER['CONTENT_TYPE'] ?? ''; + self::parseMultipart($origCt, $rawBody, $_POST, $_FILES); } - $_REQUEST = array_merge($_GET, $_POST); + $_REQUEST = array_merge($_COOKIE, $_GET, $_POST); // PHP default order + + // Make raw body available + Q_Request::$input = $rawBody; // Clear any stale headers and output from previous in-process requests, // then start fresh output buffering. This prevents "headers already sent" @@ -1475,7 +1762,7 @@ HTML } $body = ob_get_clean(); header_remove(); - list($_SERVER, $_GET, $_POST, $_REQUEST) = $saved; + list($_SERVER, $_GET, $_POST, $_REQUEST, $_COOKIE) = $saved; // Process Merkle cache headers (strips X-Q-Cache-* from response) if (Q_WebServer_Cache_Components::enabled()) { @@ -1546,6 +1833,114 @@ HTML return compact('method', 'uri', 'path', 'query', 'headers', 'body', 'httpVersion'); } + /** + * Parse multipart/form-data body into $_POST and $_FILES arrays. + * Handles file uploads by writing to temp files (same as php-fpm). + * @method parseMultipart + * @static + * @param {string} $contentType Full Content-Type header value + * @param {string} $body Raw request body + * @param {array} &$post Populated with form field values + * @param {array} &$files Populated with file upload entries + */ + static function parseMultipart($contentType, $body, &$post, &$files) + { + // Extract boundary from Content-Type + if (!preg_match('/boundary=(?:"([^"]+)"|([^\s;]+))/i', $contentType, $bm)) { + return; + } + $boundary = '--' . ($bm[1] ?: $bm[2]); + $endBoundary = $boundary . '--'; + + $parts = explode($boundary, $body); + array_shift($parts); // before first boundary + + foreach ($parts as $part) { + $part = ltrim($part, "\r\n"); + if ($part === '--' || $part === "--\r\n" || $part === '') continue; + if (strpos($part, '--') === 0) continue; // end boundary + + // Split headers from body + $headerEnd = strpos($part, "\r\n\r\n"); + if ($headerEnd === false) continue; + + $headerBlock = substr($part, 0, $headerEnd); + $partBody = substr($part, $headerEnd + 4); + // Remove trailing \r\n + if (substr($partBody, -2) === "\r\n") { + $partBody = substr($partBody, 0, -2); + } + + // Parse part headers + $partHeaders = array(); + foreach (explode("\r\n", $headerBlock) as $line) { + $colonPos = strpos($line, ':'); + if ($colonPos !== false) { + $k = strtolower(trim(substr($line, 0, $colonPos))); + $v = trim(substr($line, $colonPos + 1)); + $partHeaders[$k] = $v; + } + } + + $disp = $partHeaders['content-disposition'] ?? ''; + if (strpos($disp, 'form-data') === false) continue; + + // Extract name + $name = null; + if (preg_match('/\bname="([^"]*)"/', $disp, $nm)) { + $name = $nm[1]; + } elseif (preg_match("/\bname='([^']*)'/", $disp, $nm)) { + $name = $nm[1]; + } + if ($name === null) continue; + + // Check if it's a file upload + $filename = null; + if (preg_match('/\bfilename="([^"]*)"/', $disp, $fm)) { + $filename = $fm[1]; + } elseif (preg_match("/\bfilename='([^']*)'/", $disp, $fm)) { + $filename = $fm[1]; + } + + if ($filename !== null) { + // File upload β€” write to temp file + $tmpPath = tempnam(sys_get_temp_dir(), 'qbix_upload_'); + file_put_contents($tmpPath, $partBody); + + $fileEntry = array( + 'name' => $filename, + 'type' => $partHeaders['content-type'] ?? 'application/octet-stream', + 'tmp_name' => $tmpPath, + 'error' => UPLOAD_ERR_OK, + 'size' => strlen($partBody), + ); + + // Handle array notation: files[0], files[photo], etc. + if (preg_match('/^([^\[]+)\[([^\]]*)\]$/', $name, $am)) { + $files[$am[1]]['name'][$am[2]] = $fileEntry['name']; + $files[$am[1]]['type'][$am[2]] = $fileEntry['type']; + $files[$am[1]]['tmp_name'][$am[2]] = $fileEntry['tmp_name']; + $files[$am[1]]['error'][$am[2]] = $fileEntry['error']; + $files[$am[1]]['size'][$am[2]] = $fileEntry['size']; + } else { + $files[$name] = $fileEntry; + } + } else { + // Regular form field + // Handle array notation: tags[], data[key], etc. + if (preg_match('/^([^\[]+)\[([^\]]*)\]$/', $name, $am)) { + if ($am[2] === '') { + $post[$am[1]][] = $partBody; + } else { + $post[$am[1]][$am[2]] = $partBody; + } + } else { + $post[$name] = $partBody; + } + } + } + } + // ── Response helpers ───────────────────────────────── static function sendResponse($client, $status, $body, $type = 'text/plain; charset=utf-8', $extra = array()) diff --git a/src/Q/WebServer/Pool.php b/src/Q/WebServer/Pool.php index 4fd0f77..cb9ab55 100644 --- a/src/Q/WebServer/Pool.php +++ b/src/Q/WebServer/Pool.php @@ -361,7 +361,7 @@ class Q_WebServer_Pool // Send SIGTERM to all workers foreach ($this->workers as $w) { - posix_kill($w['pid'], SIGTERM); + if (function_exists("posix_kill")) posix_kill($w["pid"], SIGTERM); } // Wait for workers to exit gracefully @@ -381,7 +381,7 @@ class Q_WebServer_Pool // SIGKILL any workers that didn't exit in time foreach ($remaining as $w) { - posix_kill($w['pid'], SIGKILL); + if (function_exists("posix_kill")) posix_kill($w["pid"], SIGKILL); pcntl_waitpid($w['pid'], $st, 0); } diff --git a/src/Q/WebSocket.php b/src/Q/WebSocket.php index a618dde..e97ac2e 100644 --- a/src/Q/WebSocket.php +++ b/src/Q/WebSocket.php @@ -213,6 +213,8 @@ class Q_WebSocket 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 => $_) { @@ -300,4 +302,253 @@ class Q_WebSocket $frame .= $payload; @fwrite($socket, $frame); } + + // ── Process-per-socket dispatch ───────────────────── + + /** + * 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 + if (!isset(self::$workers[$socketKey])) { + self::spawnWorker($socketKey, $path); + } + + if (!isset(self::$workers[$socketKey])) return; // fork failed + + // Forward message to child via length-prefixed JSON + $json = json_encode($msg, JSON_UNESCAPED_SLASHES); + $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])) { + @fwrite(self::$workers[$socketKey]['pipe'], $packet); + } + } + } + + /** + * 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 + } + + $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 === 0) { + // ── CHILD: 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( + '_socketId' => $socketKey, '_path' => $path, + 'event' => '_connect', 'data' => array(), + )); + 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; + $params = array( + '_socketId' => $socketKey, + '_path' => $path, + '_ack' => Q_Socket::$_ack, + '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( + '_socketId' => $socketKey, 'event' => '_disconnect', 'data' => array(), + )); + 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 ($socketKey) { + $data = @fread($pipe, 65536); + if ($data === false || $data === '') { + Q_WebSocket::cleanupWorker($socketKey); + 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::$workers[$socketKey] = array( + '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); + 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); + } + + /** + * Run a socket event handler in-process (Windows/no fork fallback). + * @method dispatchEventInProcess + * @static + */ + 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 + */ + static function executeCommand($cmd) + { + switch ($cmd['cmd'] ?? '') { + case 'send': + self::send($cmd['socketId'], $cmd['data']); + break; + case 'broadcast': + self::broadcastTo($cmd['room'], $cmd['data']); + break; + case 'broadcastAll': + self::broadcast($cmd['data']); + break; + case 'join': + self::subscribe($cmd['socketId'], $cmd['room']); + break; + case 'leave': + self::unsubscribe($cmd['socketId'], $cmd['room']); + break; + } + } } diff --git a/test/web/qbix-socket.js b/test/web/qbix-socket.js new file mode 100644 index 0000000..708775d --- /dev/null +++ b/test/web/qbix-socket.js @@ -0,0 +1,142 @@ +/** + * QSocket β€” tiny WebSocket client for Qbix Server. + * + * Usage: + * var qs = new QSocket('ws://localhost:8080/ws/chat'); + * + * qs.on('chat/message', function(data) { + * console.log(data.from + ': ' + data.text); + * }); + * + * qs.emit('chat/message', {text: 'hello'}, function(ack) { + * console.log('Server confirmed:', ack); + * }); + * + * qs.emit('chat/join', {room: 'lobby'}); + * + * Protocol (JSON over WebSocket): + * Client β†’ Server: {"event": "...", "data": {...}, "ack": N} + * Server β†’ Client: {"event": "...", "data": {...}} (broadcast) + * Server β†’ Client: {"ack": N, "data": {...}} (callback) + */ +(function (root) { + 'use strict'; + + function QSocket(url, options) { + var self = this; + options = options || {}; + self._handlers = {}; + self._ackId = 0; + self._acks = {}; + self._queue = []; + self._reconnect = options.reconnect !== false; + self._reconnectDelay = options.reconnectDelay || 1000; + self._maxDelay = options.maxReconnectDelay || 30000; + self._url = url; + + self._connect = function () { + self.ws = new WebSocket(url); + + 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.ws.onmessage = function (e) { + var msg; + try { msg = JSON.parse(e.data); } catch (err) { return; } + + // Ack response (callback from server) + 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); + }); + } + }; + + self.ws.onclose = function () { + if (self._handlers['disconnect']) { + self._handlers['disconnect'].forEach(function (fn) { fn(); }); + } + if (self._reconnect) { + self._delay = Math.min(self._delay * 1.5, self._maxDelay); + setTimeout(self._connect, self._delay); + } + }; + + self.ws.onerror = function () { + self.ws.close(); + }; + }; + + self._delay = self._reconnectDelay; + self._connect(); + } + + /** + * Listen for an event from the server. + * Special events: 'connect', 'disconnect' + */ + QSocket.prototype.on = function (event, fn) { + if (!this._handlers[event]) this._handlers[event] = []; + this._handlers[event].push(fn); + return this; + }; + + /** + * Remove a listener. + */ + 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; }); + return this; + }; + + /** + * Send an event to the server. + * Optional callback is invoked when the server acks. + */ + QSocket.prototype.emit = function (event, data, callback) { + var msg = { event: event, data: data || {} }; + if (typeof callback === 'function') { + msg.ack = ++this._ackId; + this._acks[msg.ack] = callback; + } + var json = JSON.stringify(msg); + if (this.ws && this.ws.readyState === WebSocket.OPEN) { + this.ws.send(json); + } else { + this._queue.push(json); + } + return this; + }; + + /** + * Close the connection (disables auto-reconnect). + */ + QSocket.prototype.close = function () { + this._reconnect = false; + if (this.ws) this.ws.close(); + }; + + // Export + if (typeof module !== 'undefined' && module.exports) { + module.exports = QSocket; + } else { + root.QSocket = QSocket; + } +})(typeof window !== 'undefined' ? window : this); diff --git a/web/qbix-socket.js b/web/qbix-socket.js new file mode 100644 index 0000000..708775d --- /dev/null +++ b/web/qbix-socket.js @@ -0,0 +1,142 @@ +/** + * QSocket β€” tiny WebSocket client for Qbix Server. + * + * Usage: + * var qs = new QSocket('ws://localhost:8080/ws/chat'); + * + * qs.on('chat/message', function(data) { + * console.log(data.from + ': ' + data.text); + * }); + * + * qs.emit('chat/message', {text: 'hello'}, function(ack) { + * console.log('Server confirmed:', ack); + * }); + * + * qs.emit('chat/join', {room: 'lobby'}); + * + * Protocol (JSON over WebSocket): + * Client β†’ Server: {"event": "...", "data": {...}, "ack": N} + * Server β†’ Client: {"event": "...", "data": {...}} (broadcast) + * Server β†’ Client: {"ack": N, "data": {...}} (callback) + */ +(function (root) { + 'use strict'; + + function QSocket(url, options) { + var self = this; + options = options || {}; + self._handlers = {}; + self._ackId = 0; + self._acks = {}; + self._queue = []; + self._reconnect = options.reconnect !== false; + self._reconnectDelay = options.reconnectDelay || 1000; + self._maxDelay = options.maxReconnectDelay || 30000; + self._url = url; + + self._connect = function () { + self.ws = new WebSocket(url); + + 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.ws.onmessage = function (e) { + var msg; + try { msg = JSON.parse(e.data); } catch (err) { return; } + + // Ack response (callback from server) + 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); + }); + } + }; + + self.ws.onclose = function () { + if (self._handlers['disconnect']) { + self._handlers['disconnect'].forEach(function (fn) { fn(); }); + } + if (self._reconnect) { + self._delay = Math.min(self._delay * 1.5, self._maxDelay); + setTimeout(self._connect, self._delay); + } + }; + + self.ws.onerror = function () { + self.ws.close(); + }; + }; + + self._delay = self._reconnectDelay; + self._connect(); + } + + /** + * Listen for an event from the server. + * Special events: 'connect', 'disconnect' + */ + QSocket.prototype.on = function (event, fn) { + if (!this._handlers[event]) this._handlers[event] = []; + this._handlers[event].push(fn); + return this; + }; + + /** + * Remove a listener. + */ + 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; }); + return this; + }; + + /** + * Send an event to the server. + * Optional callback is invoked when the server acks. + */ + QSocket.prototype.emit = function (event, data, callback) { + var msg = { event: event, data: data || {} }; + if (typeof callback === 'function') { + msg.ack = ++this._ackId; + this._acks[msg.ack] = callback; + } + var json = JSON.stringify(msg); + if (this.ws && this.ws.readyState === WebSocket.OPEN) { + this.ws.send(json); + } else { + this._queue.push(json); + } + return this; + }; + + /** + * Close the connection (disables auto-reconnect). + */ + QSocket.prototype.close = function () { + this._reconnect = false; + if (this.ws) this.ws.close(); + }; + + // Export + if (typeof module !== 'undefined' && module.exports) { + module.exports = QSocket; + } else { + root.QSocket = QSocket; + } +})(typeof window !== 'undefined' ? window : this);