Added support for socket.io and simple JSON websocket payloads via /Q/socket.js

qbixserver.php

@@ -205,6 +205,9 @@ if (file_exists($appConfig)) {
 	Q_Config::load($appConfig);
 }
 
+// Preload handlers if configured (Q.handlers.preload: true)
+Q::preload();
+
 // ── PID file ────────────────────────────────────────
 
 if ($opts['pid']) {
@@ -242,6 +245,12 @@ echo "  │" . str_pad("  http://{$opts['host']}:{$opts['port']}", $W) . "│\n"
 echo "  │" . str_pad("  Root: " . basename($webDir), $W) . "│\n";
 echo "  │" . str_pad("  Mode: " . ($qbixMode ? 'Qbix Platform' : 'Standalone'), $W) . "│\n";
 echo "  │" . str_pad("  PHP: " . ($opts['workers'] ? $opts['workers'] . ' workers' : 'in-process'), $W) . "│\n";
+$nClasses = count(get_declared_classes());
+$nHandlers = Q::$preloadedHandlers;
+$preloadLabel = $nHandlers > 0
+	? "  Preloaded: {$nClasses} classes, {$nHandlers} handlers"
+	: "  Preloaded: {$nClasses} classes (handlers: lazy)";
+echo "  │" . str_pad($preloadLabel, $W) . "│\n";
 echo "  ├" . str_repeat('─', $W) . "┤\n";
 echo "  │" . str_pad("  Dashboard: /Q/dashboard", $W) . "│\n";
 echo "  │" . str_pad("  Health:    /Q/health", $W) . "│\n";

src/Q.php

@@ -137,6 +137,18 @@ class Q
 	static $_responseHeaders = array();
 	static $_responseCode = 200;
 
+	/**
+	 * Get the app name. Used to prefix handler function names.
+	 * Set via config: {"Q": {"app": "MyApp"}}
+	 * @method app
+	 * @static
+	 * @return {string} App name, or empty string if not set
+	 */
+	static function app()
+	{
+		return Q_Config::get('Q', 'app', '');
+	}
+
 	/**
 	 * Set a response header. Wraps PHP's header() and captures it.
 	 * Scripts can call either header() directly or Q::header() — both work.
@@ -287,7 +299,9 @@ class Q
 	static function canHandle($eventName)
 	{
 		$parts = explode('/', $eventName);
-		$funcName = str_replace('-', '_', implode('_', $parts));
+		$baseName = str_replace('-', '_', implode('_', $parts));
+		$app = Q::app();
+		$funcName = ($app !== '' ? $app . '_' : '') . $baseName;
 		if (function_exists($funcName)) return true;
 
 		// Try to load from handlers/ directory
@@ -327,7 +341,9 @@ class Q
 		}
 
 		$parts = explode('/', $eventName);
-		$funcName = str_replace('-', '_', implode('_', $parts));
+		$baseName = str_replace('-', '_', implode('_', $parts));
+		$app = Q::app();
+		$funcName = ($app !== '' ? $app . '_' : '') . $baseName;
 
 		if (!function_exists($funcName)) {
 			if ($skipIncludes) return null;
@@ -481,6 +497,57 @@ class Q
 			self::$paths[] = $projectRoot;
 		}
 	}
+
+	/**
+	 * Preload all handler files if Q.handlers.preload is true.
+	 * Call this after config is loaded and before the server starts accepting
+	 * connections. Handlers are included once in the parent process and shared
+	 * via COW across all forked children.
+	 *
+	 * Off by default — handlers lazy-load via include_once on first call,
+	 * which is fine with opcache (edit a file, refresh, see the change).
+	 * Enable in production for full COW sharing of handler bytecode.
+	 *
+	 * @method preload
+	 * @static
+	 */
+	static function preload()
+	{
+		if (!Q_Config::get('Q', 'handlers', 'preload', false)) {
+			return;
+		}
+		foreach (self::$paths as $base) {
+			$handlersDir = $base . DS . 'handlers';
+			if (is_dir($handlersDir)) {
+				self::preloadDir($handlersDir);
+			}
+		}
+	}
+
+	/**
+	 * Recursively include all .php files in a directory.
+	 * @method preloadDir
+	 * @static
+	 * @param {string} $dir Directory to scan
+	 */
+	static function preloadDir($dir)
+	{
+		$entries = @scandir($dir);
+		if (!$entries) return;
+		foreach ($entries as $entry) {
+			if ($entry[0] === '.') continue;
+			$path = $dir . DS . $entry;
+			if (is_dir($path)) {
+				self::preloadDir($path);
+			} elseif (substr($entry, -4) === '.php') {
+				include_once $path;
+				self::$preloadedHandlers++;
+			}
+		}
+	}
+
+	/** @var integer Number of preloaded handler files */
+	static $preloadedHandlers = 0;
 }
 
 spl_autoload_register(array('Q', 'autoload'));
@@ -488,80 +555,117 @@ spl_autoload_register(array('Q', 'autoload'));
 // ── Q_Socket ────────────────────────────────────────
 
 /**
- * PHP API for WebSocket handlers — sending messages, managing rooms.
+ * WebSocket connection context. Passed to per-connection handlers as
+ * $params['socket']. Use instance methods to communicate with clients.
  *
- * Each WebSocket connection gets one PHP process. The server dispatches
+ *   function my_handler(&$params, &$result) {
- * messages via Q::event() to handlers. Handlers use Q_Socket to send
+ *       extract($params); // $socket, $event, $data
- * data back. Static variables in handlers persist across messages
+ *       $socket->reply(['hello' => 'world']);
- * (same process) and are wiped on disconnect (process dies).
+ *       $socket->join('chat/general', ['name' => 'Alice']);
+ *       $location = $socket->getLocation(); // RPC call to client
+ *   }
  *
  * @class Q_Socket
  */
 class Q_Socket
 {
-	/** @var resource IPC pipe to parent */
+	/** @var integer This socket's ID */
-	static $_pipe = null;
+	public $id;
-	/** @var integer Current client's socket key */
+
-	static $_socketId = null;
+	function __construct($id) { $this->id = $id; }
-	/** @var integer|null Ack ID from current message */
+
-	static $_ack = null;
+	/** Get a socket instance by ID */
-	/** @var boolean True when running in-process (no fork) */
+	static function byId($id) { return new self($id); }
-	static $_directMode = false;
+
-	/** @var array Buffered outbound commands */
+	/** Send data to this socket's client */
-	static $_buffer = array();
+	function reply($data) { self::_cmd(array('cmd' => 'send', 'socketId' => $this->id, 'data' => $data)); }
+
+	/** Send data to a specific client by socket ID */
+	function send($socketId, $data) { self::_cmd(array('cmd' => 'send', 'socketId' => $socketId, 'data' => $data)); }
+
+	/** Broadcast to all clients in a room */
+	function broadcast($room, $data) { self::_cmd(array('cmd' => 'broadcast', 'room' => $room, 'data' => $data)); }
+
+	/** Broadcast to ALL connected clients */
+	function broadcastAll($data) { self::_cmd(array('cmd' => 'broadcastAll', 'data' => $data)); }
+
+	/** Join a room, optionally forwarding data to the room's join handler */
+	function join($room, $data = array()) { self::_cmd(array('cmd' => 'join', 'socketId' => $this->id, 'room' => $room, 'data' => $data)); }
+
+	/** Leave a room, optionally forwarding data to the room's leave handler */
+	function leave($room, $data = array()) { self::_cmd(array('cmd' => 'leave', 'socketId' => $this->id, 'room' => $room, 'data' => $data)); }
 
 	/**
-	 * Send data to the client that owns this connection.
+	 * Call a method on the remote client. Blocks until the client responds.
+	 * The client must have registered a handler via qs.handle('methodName', fn).
+	 *
+	 * @method __call
+	 * @param {string} $method Method name to invoke on the client
+	 * @param {array} $args Arguments — first element is passed as data to client
+	 * @return {mixed} Return value from the client handler, or null on timeout
 	 */
-	static function reply($data)
+	function __call($method, $args)
 	{
-		self::send(self::$_socketId, $data);
+		$rpcId = ++self::$_rpcCounter;
+		$data = isset($args[0]) ? $args[0] : array();
+
+		// Flush any pending commands first
+		self::flush();
+
+		// Write RPC request directly to pipe (not buffered — need immediate send)
+		$cmd = json_encode(array(
+			'cmd' => 'rpc', 'socketId' => $this->id,
+			'method' => $method, 'data' => $data, 'rpcId' => $rpcId,
+		), JSON_UNESCAPED_SLASHES) . "\n";
+		@fwrite(self::$_pipe, $cmd);
+
+		// Block reading pipe until we get our RPC response (timeout 5s)
+		$deadline = microtime(true) + 5.0;
+		while (microtime(true) < $deadline) {
+			$remaining = $deadline - microtime(true);
+			if ($remaining <= 0) break;
+
+			$read = array(self::$_pipe);
+			$w = $e = null;
+			$sec = (int) $remaining;
+			$usec = (int) (($remaining - $sec) * 1000000);
+			if (@stream_select($read, $w, $e, $sec, $usec) < 1) break;
+
+			$header = @fread(self::$_pipe, 4);
+			if (!$header || strlen($header) < 4) break;
+			$len = unpack('N', $header)[1];
+			if ($len <= 0 || $len > 10485760) break;
+			$json = '';
+			while (strlen($json) < $len) {
+				$chunk = @fread(self::$_pipe, $len - strlen($json));
+				if ($chunk === false || $chunk === '') break 2;
+				$json .= $chunk;
+			}
+			$msg = json_decode($json, true);
+			if (!$msg) continue;
+
+			// Is this our RPC response?
+			if (isset($msg['_rpc']) && $msg['_rpc'] === $rpcId) {
+				return isset($msg['result']) ? $msg['result'] : null;
 			}
 
-	/**
+			// Not our response — buffer for the main loop
-	 * Send data to a specific connected client.
+			self::$_messageQueue[] = $msg;
-	 */
+		}
-	static function send($socketId, $data)
+		return null; // timeout
-	{
-		self::_command(array('cmd' => 'send', 'socketId' => $socketId, 'data' => $data));
 	}
 
-	/**
+	// ── Internal IPC plumbing (not part of the public API) ──
-	 * Broadcast to all clients in a room/channel.
-	 */
-	static function broadcast($room, $data)
-	{
-		self::_command(array('cmd' => 'broadcast', 'room' => $room, 'data' => $data));
-	}
 
-	/**
+	/** @internal */ static $_pipe = null;
-	 * Broadcast to ALL connected WebSocket clients.
+	/** @internal */ static $_ack = null;
-	 */
+	/** @internal */ static $_directMode = false;
-	static function broadcastAll($data)
+	/** @internal */ static $_buffer = array();
-	{
+	/** @internal */ static $_rpcCounter = 0;
-		self::_command(array('cmd' => 'broadcastAll', 'data' => $data));
+	/** @internal */ static $_messageQueue = array();
-	}
 
-	/**
+	/** @internal */
-	 * Subscribe a client to a room/channel.
+	static function _cmd($cmd)
-	 */
-	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);
@@ -570,10 +674,7 @@ class Q_Socket
 		}
 	}
 
-	/**
+	/** @internal Flush buffered commands to IPC pipe */
-	 * Flush buffered commands to the IPC pipe.
-	 * Called automatically after each handler invocation.
-	 */
 	static function flush()
 	{
 		if (!self::$_pipe || empty(self::$_buffer)) return;
@@ -586,6 +687,48 @@ class Q_Socket
 	}
 }
 
+// ── Q_Room ──────────────────────────────────────────
+
+/**
+ * Room context. Passed to room handlers as $params['room'].
+ * Wraps IPC commands with room context for cleaner handler code.
+ *
+ *   function chat_room_message(&$params, &$result) {
+ *       extract($params); // $room, $event, $data
+ *       $room->broadcast(['event' => 'chat/message', 'data' => $data]);
+ *   }
+ *
+ * @class Q_Room
+ */
+class Q_Room
+{
+	/** @var string Room name (e.g. 'chat/general') */
+	public $name;
+	/** @var integer Socket ID of the current message sender (0 for lifecycle events without a sender) */
+	public $socketId;
+	/** @var array Pattern params (e.g. ['room' => 'general'] from 'chat/$room') */
+	public $params;
+
+	function __construct($name, $socketId = 0, $params = array())
+	{
+		$this->name = $name;
+		$this->socketId = $socketId;
+		$this->params = $params;
+	}
+
+	/** Get a room instance by name */
+	static function byName($name) { return new self($name); }
+
+	/** Send to all members in this room */
+	function broadcast($data) { Q_Socket::_cmd(array('cmd' => 'broadcast', 'room' => $this->name, 'data' => $data)); }
+
+	/** Send to the member who sent the current message */
+	function reply($data) { Q_Socket::_cmd(array('cmd' => 'send', 'socketId' => $this->socketId, 'data' => $data)); }
+
+	/** Send to a specific member by socket ID */
+	function send($socketId, $data) { Q_Socket::_cmd(array('cmd' => 'send', 'socketId' => $socketId, 'data' => $data)); }
+}
+
 // ── Q_Request ───────────────────────────────────────
 
 /**

src/Q/WebServer.php

@@ -750,6 +750,25 @@ class Q_WebServer
 		$path = $parsed['path'];
 
 		// 1. Dashboard + Panel + WebSocket + Health (/Q/*)
+		// 1. Serve client JS files (before /Q/ check — socket.io.js is at /socket.io/)
+		$jsMap = array();
+		$jsPath = Q_Config::get('Q', 'socket', 'js', '/Q/socket.js');
+		if ($jsPath !== false) $jsMap[$jsPath] = __DIR__ . DS . 'socket.js';
+		$ioPath = Q_Config::get('Q', 'socket', 'io', '/socket.io');
+		if ($ioPath !== false) $jsMap[$ioPath . '/socket.io.js'] = __DIR__ . DS . 'socket.io.js';
+		if (isset($jsMap[$path])) {
+			$jsFile = $jsMap[$path];
+			if (file_exists($jsFile)) {
+				self::sendResponse($client, 200, file_get_contents($jsFile),
+					'application/javascript',
+					array('Cache-Control' => 'public, max-age=3600'));
+			} else {
+				self::sendResponse($client, 404, 'Not found');
+			}
+			return false;
+		}
+
+		// 2. Dashboard + Panel + WebSocket + Health (/Q/*)
 		if (strpos($path, '/Q/') === 0) {
 			if ($path === '/Q/ws') {
 				$upgraded = Q_WebSocket::upgrade(
@@ -779,7 +798,8 @@ class Q_WebServer
 				$client, $parsed['headers'],
 				function ($sk, $msg) use ($path) {
 					Q_WebSocket::dispatchEvent($sk, $msg, $path);
-				}
+				},
+				null, $path
 			);
 			return $upgraded;
 		}

src/Q/WebSocket.php

@@ -32,7 +32,7 @@ class Q_WebSocket
 
 	// ── Upgrade + framing (unchanged) ───────────────
 
-	static function upgrade($socket, $headers, $onMessage = null, $channel = null)
+	static function upgrade($socket, $headers, $onMessage = null, $channel = null, $path = '/')
 	{
 		$key = $headers['sec-websocket-key'] ?? null;
 		if (!$key) return false;
@@ -46,10 +46,31 @@ class Q_WebSocket
 		$watcher = Q_Evented::onReadable($socket, function ($sock) use ($sk) {
 			Q_WebSocket::onData($sk, $sock);
 		});
+
+		// Socket.IO clients connect to configured path (default /socket.io)
+		// Set Q.socket.io to false to disable Socket.IO protocol
+		$ioPath = Q_Config::get('Q', 'socket', 'io', '/socket.io');
+		$proto = ($ioPath !== false && strpos($path, $ioPath) === 0) ? 'socketio' : 'json';
+
 		self::$clients[$sk] = array(
 			'socket' => $socket, 'watcher' => $watcher,
-			'channels' => array(), 'buffer' => '', 'onMessage' => $onMessage
+			'channels' => array(), 'buffer' => '', 'onMessage' => $onMessage,
+			'protocol' => $proto,
 		);
+
+		// Socket.IO: send Engine.IO OPEN handshake
+		if ($proto === 'socketio') {
+			$sid = base_convert(mt_rand(1000000, 9999999) . $sk, 10, 36);
+			$handshake = '0' . json_encode(array(
+				'sid' => $sid,
+				'upgrades' => array(),
+				'pingInterval' => 25000,
+				'pingTimeout' => 20000,
+				'maxPayload' => 1000000,
+			));
+			self::sendRaw($sk, $handshake);
+		}
+
 		if ($channel) self::subscribe($sk, $channel);
 		return true;
 	}
@@ -123,44 +144,44 @@ class Q_WebSocket
 	static function send($socketKey, $data)
 	{
 		if (!isset(self::$clients[$socketKey])) return;
-		$json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
+		$encoded = self::encodeSend($socketKey, $data);
-		self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $json);
+		self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $encoded);
 	}
 
 	static function broadcast($data)
 	{
-		$json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
 		foreach (self::$clients as $sk => $c) {
-			self::encodeAndSend($c['socket'], 0x1, $json);
+			$encoded = self::encodeSend($sk, $data);
+			self::encodeAndSend($c['socket'], 0x1, $encoded);
 		}
 	}
 
 	static function broadcastTo($channel, $data)
 	{
 		if (!isset(self::$channels[$channel])) return;
-		$json = is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
 		foreach (self::$channels[$channel] as $sk => $_) {
 			if (isset(self::$clients[$sk])) {
-				self::encodeAndSend(self::$clients[$sk]['socket'], 0x1, $json);
+				$encoded = self::encodeSend($sk, $data);
+				self::encodeAndSend(self::$clients[$sk]['socket'], 0x1, $encoded);
 			}
 		}
 	}
 
-	static function subscribe($sk, $channel)
+	static function subscribe($sk, $channel, $data = array())
 	{
 		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);
+		self::notifyRoomJoin($channel, $sk, $data);
 	}
 
-	static function unsubscribe($sk, $channel)
+	static function unsubscribe($sk, $channel, $data = array())
 	{
 		unset(self::$channels[$channel][$sk]);
 		if (empty(self::$channels[$channel])) unset(self::$channels[$channel]);
 		if (isset(self::$clients[$sk])) unset(self::$clients[$sk]['channels'][$channel]);
-		self::notifyRoomLeave($channel, $sk);
+		self::notifyRoomLeave($channel, $sk, $data);
 	}
 
 	static function disconnect($sk)
@@ -196,8 +217,22 @@ class Q_WebSocket
 
 	static function dispatchEvent($socketKey, $raw, $path = '/')
 	{
+		$proto = self::$clients[$socketKey]['protocol'] ?? 'json';
+
+		if ($proto === 'socketio') {
+			$msg = self::parseSocketIO($socketKey, $raw);
+			if ($msg === null) return; // handled internally (ping/pong/connect)
+		} else {
+			// Bare WebSocket — plain JSON
 			$msg = json_decode($raw, true);
-		if (!$msg || empty($msg['event'])) return;
+			if (!$msg) return;
+			// Ack-only response (client responding to server RPC)
+			if (isset($msg['ack']) && !isset($msg['event'])) {
+				self::handleRpcResponse($msg['ack'], $msg['data'] ?? null);
+				return;
+			}
+			if (empty($msg['event'])) return;
+		}
 
 		$event = $msg['event'];
 
@@ -231,6 +266,157 @@ class Q_WebSocket
 		}
 	}
 
+	// ── Socket.IO protocol support ──────────────────
+
+	/**
+	 * Parse a Socket.IO/Engine.IO message. Returns normalized internal
+	 * format or null if the message was handled internally (ping, connect).
+	 */
+	static function parseSocketIO($socketKey, $raw)
+	{
+		if ($raw === '') return null;
+		$eioType = $raw[0];
+
+		switch ($eioType) {
+			case '2': // Engine.IO ping
+				self::sendRaw($socketKey, '3'); // pong
+				return null;
+			case '3': // Engine.IO pong
+				return null;
+			case '5': // Engine.IO upgrade
+				return null;
+			case '4': // Engine.IO message → Socket.IO packet
+				break;
+			default:
+				return null;
+		}
+
+		// Strip Engine.IO prefix "4"
+		$sio = substr($raw, 1);
+		if ($sio === '' || $sio === false) return null;
+		$sioType = $sio[0];
+		$rest = substr($sio, 1);
+
+		// Extract namespace from packet (before comma or ack digits)
+		$ns = '';
+		if (isset($rest[0]) && $rest[0] === '/') {
+			$commaPos = strpos($rest, ',');
+			if ($commaPos !== false) {
+				$ns = substr($rest, 1, $commaPos - 1); // strip leading /
+				$rest = substr($rest, $commaPos + 1);
+			}
+		}
+
+		switch ($sioType) {
+			case '0': // CONNECT to namespace
+				$sid = base_convert(mt_rand(1000000, 9999999) . microtime(true) * 1000, 10, 36);
+				$nsPrefix = $ns ? '/' . $ns . ',' : '';
+
+				// Try connect handler (optional — auto-accepts if no handler)
+				$connectEvent = $ns ? $ns . '/connect' : 'connect';
+				if (Q::canHandle($connectEvent)) {
+					// Return as event so it dispatches to the handler
+					return array('event' => $connectEvent, 'data' => array(),
+						'_ns' => $ns, '_nsConnect' => true, '_nsSid' => $sid);
+				}
+
+				// Auto-accept: send CONNECT ack
+				self::sendRaw($socketKey, '40' . $nsPrefix . '{"sid":"' . $sid . '"}');
+				// Store namespace membership
+				if (!isset(self::$clients[$socketKey]['namespaces'])) {
+					self::$clients[$socketKey]['namespaces'] = array();
+				}
+				self::$clients[$socketKey]['namespaces'][$ns] = true;
+				return null;
+
+			case '1': // DISCONNECT from namespace
+				$disconnectEvent = $ns ? $ns . '/disconnect' : 'disconnect';
+				if (isset(self::$clients[$socketKey]['namespaces'])) {
+					unset(self::$clients[$socketKey]['namespaces'][$ns]);
+				}
+				return array('event' => '_disconnect', 'data' => array(), '_ns' => $ns);
+
+			case '2': // EVENT (possibly with ack)
+				// Extract optional ack ID (digits before JSON array)
+				$ackId = null;
+				$i = 0;
+				while ($i < strlen($rest) && ctype_digit($rest[$i])) $i++;
+				if ($i > 0) {
+					$ackId = (int) substr($rest, 0, $i);
+					$rest = substr($rest, $i);
+				}
+				$arr = json_decode($rest, true);
+				if (!is_array($arr) || empty($arr)) return null;
+				$eventName = array_shift($arr);
+				$data = isset($arr[0]) ? $arr[0] : array();
+				// Prepend namespace to event name
+				if ($ns) $eventName = $ns . '/' . $eventName;
+				$msg = array('event' => $eventName, 'data' => $data);
+				if ($ackId !== null) $msg['ack'] = $ackId;
+				return $msg;
+
+			case '3': // ACK (client responding to server RPC)
+				$i = 0;
+				while ($i < strlen($rest) && ctype_digit($rest[$i])) $i++;
+				$ackId = ($i > 0) ? (int) substr($rest, 0, $i) : null;
+				$rest = substr($rest, $i);
+				$arr = json_decode($rest, true);
+				$result = (is_array($arr) && !empty($arr)) ? $arr[0] : null;
+				if ($ackId !== null) {
+					self::handleRpcResponse($ackId, $result);
+				}
+				return null;
+
+			default:
+				return null;
+		}
+	}
+
+	/**
+	 * Send raw text frame to a WebSocket client (no JSON wrapping).
+	 * Used for Socket.IO protocol frames.
+	 */
+	static function sendRaw($socketKey, $text)
+	{
+		if (!isset(self::$clients[$socketKey]['socket'])) return;
+		self::encodeAndSend(self::$clients[$socketKey]['socket'], 0x1, $text);
+	}
+
+	/**
+	 * Send a Socket.IO ACK response: 43<ackId>[data]
+	 */
+	/**
+	 * Send a Socket.IO ACK response: 43<ackId>[data]
+	 * or bare JSON: {"ack": ackId, "data": ...}
+	 */
+	static function sendAck($socketKey, $ackId, $data)
+	{
+		$proto = self::$clients[$socketKey]['protocol'] ?? 'json';
+		if ($proto === 'socketio') {
+			self::sendRaw($socketKey, '43' . $ackId . json_encode(array($data), JSON_UNESCAPED_SLASHES));
+		} else {
+			self::send($socketKey, array('ack' => $ackId, 'data' => $data));
+		}
+	}
+
+	/**
+	 * Encode outgoing data for the client's protocol.
+	 */
+	static function encodeSend($socketKey, $data)
+	{
+		$proto = self::$clients[$socketKey]['protocol'] ?? 'json';
+
+		if ($proto === 'socketio') {
+			$event = $data['event'] ?? 'message';
+			$payload = $data['data'] ?? $data;
+			$arr = array($event, $payload);
+			return '42' . json_encode($arr, JSON_UNESCAPED_SLASHES);
+		}
+
+		// Bare WebSocket — plain JSON
+		return json_encode($data, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
+	}
+
 	static function spawnWorker($socketKey, $path)
 	{
 		if (!function_exists('pcntl_fork')) return;
@@ -247,18 +433,24 @@ class Q_WebSocket
 			fclose($pair[0]);
 			$pipe = $pair[1];
 			Q_Socket::$_pipe = $pipe;
-			Q_Socket::$_socketId = $socketKey;
+
+			$socket = new Q_Socket($socketKey);
 
 			$connectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_connect', null);
 			if ($connectHandler) {
 				Q::event($connectHandler, array(
-					'_socketId' => $socketKey, '_path' => $path,
+					'socket' => $socket, 'path' => $path,
 					'event' => '_connect', 'data' => array(),
 				));
 				Q_Socket::flush();
 			}
 
 			while (true) {
+				// Check message queue first (filled by __call when it
+				// reads non-RPC messages while waiting for a response)
+				if (!empty(Q_Socket::$_messageQueue)) {
+					$msg = array_shift(Q_Socket::$_messageQueue);
+				} else {
 					$header = @fread($pipe, 4);
 					if ($header === false || $header === '' || strlen($header) < 4) break;
 					$len = unpack('N', $header)[1];
@@ -271,6 +463,7 @@ class Q_WebSocket
 					}
 					$msg = json_decode($json, true);
 					if (!$msg) continue;
+				}
 
 				$event = $msg['event'] ?? '';
 				if ($event === '_disconnect') break;
@@ -280,16 +473,18 @@ class Q_WebSocket
 
 				$result = null;
 				$params = array(
-					'_socketId' => $socketKey,
+					'socket' => $socket,
-					'_path'     => $path,
+					'path'   => $path,
-					'_ack'      => Q_Socket::$_ack,
 					'event'  => $event,
 					'data'   => $msg['data'] ?? array(),
 				);
 				Q::event($mapped, $params, false, false, $result);
 
 				if (Q_Socket::$_ack !== null && $result !== null) {
-					Q_Socket::reply(array('ack' => Q_Socket::$_ack, 'data' => $result));
+					Q_Socket::_cmd(array(
+						'cmd' => 'ack', 'socketId' => $socket->id,
+						'ackId' => Q_Socket::$_ack, 'data' => $result,
+					));
 				}
 				Q_Socket::flush();
 			}
@@ -297,7 +492,7 @@ class Q_WebSocket
 			$disconnectHandler = Q_Config::get('Q', 'webserver', 'sockets', 'events', '_disconnect', null);
 			if ($disconnectHandler) {
 				Q::event($disconnectHandler, array(
-					'_socketId' => $socketKey, 'event' => '_disconnect', 'data' => array(),
+					'socket' => $socket, 'event' => '_disconnect', 'data' => array(),
 				));
 				Q_Socket::flush();
 			}
@@ -423,7 +618,6 @@ class Q_WebSocket
 			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;
@@ -431,20 +625,20 @@ class Q_WebSocket
 				$tickCallback = function () use ($handler, $roomName, $params, $pipe) {
 					Q_Socket::$_ack = null;
 					$result = null;
+					$room = new Q_Room($roomName, 0, $params);
 					$p = array_merge($params, array(
-						'_room' => $roomName, 'event' => '_tick',
+						'room' => $room, 'event' => '_tick', 'data' => array(),
-						'data' => array(), '_socketId' => 0,
 					));
-					Q::event($handler, $p, false, false, $result);
+					Q::event($handler . '/tick', $p, false, false, $result);
 					Q_Socket::flush();
 				};
 			}
 
 			// Fire _init event
 			$result = null;
-			Q::event($handler, array_merge($params, array(
+			$room = new Q_Room($roomName, 0, $params);
-				'_room' => $roomName, 'event' => '_init', 'data' => array(),
+			Q::event($handler . '/init', array_merge($params, array(
-				'_socketId' => 0,
+				'room' => $room, 'event' => '_init', 'data' => array(),
 			)), false, false, $result);
 			Q_Socket::flush();
 
@@ -488,30 +682,34 @@ class Q_WebSocket
 					if ($event === '_shutdown') break 2;
 
 					Q_Socket::$_ack = isset($msg['ack']) ? $msg['ack'] : null;
-					Q_Socket::$_socketId = $msg['_socketId'] ?? 0;
+					$senderSocketId = $msg['_socketId'] ?? 0;
 
 					$result = null;
+					$room = new Q_Room($roomName, $senderSocketId, $params);
 					$p = array_merge($params, array(
-						'_room'     => $roomName,
+						'room'  => $room,
-						'_socketId' => Q_Socket::$_socketId,
-						'_ack'      => Q_Socket::$_ack,
 						'event' => $event,
 						'data'  => $msg['data'] ?? array(),
 					));
-					Q::event($handler, $p, false, false, $result);
+					// Lifecycle events: _join → handler/join
+					// User events: message → handler/message
+					$eventPath = $handler . '/' . ltrim($event, '_');
+					Q::event($eventPath, $p, false, false, $result);
 
 					if (Q_Socket::$_ack !== null && $result !== null) {
-						Q_Socket::send(Q_Socket::$_socketId,
+						Q_Socket::_cmd(array(
-							array('ack' => Q_Socket::$_ack, 'data' => $result));
+							'cmd' => 'ack', 'socketId' => $room->socketId,
+							'ackId' => Q_Socket::$_ack, 'data' => $result,
+						));
 					}
 					Q_Socket::flush();
 				}
 			}
 
 			// Fire _destroy event
-			Q::event($handler, array_merge($params, array(
+			$room = new Q_Room($roomName, 0, $params);
-				'_room' => $roomName, 'event' => '_destroy', 'data' => array(),
+			Q::event($handler . '/destroy', array_merge($params, array(
-				'_socketId' => 0,
+				'room' => $room, 'event' => '_destroy', 'data' => array(),
 			)), false, false, $result);
 			Q_Socket::flush();
 
@@ -559,7 +757,7 @@ class Q_WebSocket
 	 * @method notifyRoomJoin
 	 * @static
 	 */
-	static function notifyRoomJoin($channel, $socketKey)
+	static function notifyRoomJoin($channel, $socketKey, $data = array())
 	{
 		$config = self::matchRoomPattern($channel);
 		if (!$config) return;
@@ -572,7 +770,7 @@ class Q_WebSocket
 
 		self::$roomWorkers[$channel]['members'][$socketKey] = true;
 		self::sendToRoomWorker($channel, array(
-			'event' => '_join', 'data' => array(),
+			'event' => '_join', 'data' => $data,
 			'_socketId' => $socketKey,
 		));
 	}
@@ -582,13 +780,13 @@ class Q_WebSocket
 	 * @method notifyRoomLeave
 	 * @static
 	 */
-	static function notifyRoomLeave($channel, $socketKey)
+	static function notifyRoomLeave($channel, $socketKey, $data = array())
 	{
 		if (!isset(self::$roomWorkers[$channel])) return;
 		unset(self::$roomWorkers[$channel]['members'][$socketKey]);
 
 		self::sendToRoomWorker($channel, array(
-			'event' => '_leave', 'data' => array(),
+			'event' => '_leave', 'data' => $data,
 			'_socketId' => $socketKey,
 		));
 
@@ -624,8 +822,9 @@ class Q_WebSocket
 	static function dispatchEventInProcess($eventName, $params, $socketKey, $ack)
 	{
 		Q_Socket::$_directMode = true;
-		Q_Socket::$_socketId = $socketKey;
 		Q_Socket::$_ack = $ack;
+		$socket = new Q_Socket($socketKey);
+		$params['socket'] = $socket;
 		$result = null;
 		Q::event($eventName, $params, false, false, $result);
 		if ($ack !== null && $result !== null) {
@@ -636,6 +835,11 @@ class Q_WebSocket
 
 	// ── IPC command execution ───────────────────────
 
+	static function clientCount()
+	{
+		return count(self::$clients);
+	}
+
 	static function executeCommand($cmd)
 	{
 		switch ($cmd['cmd'] ?? '') {
@@ -649,11 +853,86 @@ class Q_WebSocket
 				self::broadcast($cmd['data']);
 				break;
 			case 'join':
-				self::subscribe($cmd['socketId'], $cmd['room']);
+				self::subscribe($cmd['socketId'], $cmd['room'], $cmd['data'] ?? array());
 				break;
 			case 'leave':
-				self::unsubscribe($cmd['socketId'], $cmd['room']);
+				self::unsubscribe($cmd['socketId'], $cmd['room'], $cmd['data'] ?? array());
+				break;
+			case 'ack':
+				self::sendAck($cmd['socketId'], $cmd['ackId'], $cmd['data']);
+				break;
+			case 'rpc':
+				self::handleRpc($cmd);
 				break;
 		}
 	}
+
+	// ── Server→Client RPC ───────────────────────────
+
+	/** @internal Maps rpcAckId → ['pipe' => resource, 'rpcId' => int] */
+	static $pendingRpc = array();
+	/** @internal Counter for server→client ack IDs */
+	static $rpcAckCounter = 0;
+
+	/**
+	 * Handle an RPC request from a child process.
+	 * Sends the method call to the client with an ack ID, then routes
+	 * the client's ack response back to the child's IPC pipe.
+	 */
+	static function handleRpc($cmd)
+	{
+		$socketKey = $cmd['socketId'];
+		$method = $cmd['method'];
+		$data = $cmd['data'] ?? array();
+		$rpcId = $cmd['rpcId'];
+
+		// Generate a unique ack ID for server→client
+		$ackId = ++self::$rpcAckCounter;
+
+		// Find which child pipe to route the response back to
+		$childPipe = null;
+		if (isset(self::$workers[$socketKey])) {
+			$childPipe = self::$workers[$socketKey]['pipe'];
+		}
+		if (!$childPipe) return;
+
+		// Store mapping so we can route the ack response back
+		self::$pendingRpc[$ackId] = array(
+			'pipe' => $childPipe,
+			'rpcId' => $rpcId,
+		);
+
+		// Send RPC call to client with ack ID
+		$proto = self::$clients[$socketKey]['protocol'] ?? 'json';
+		if ($proto === 'socketio') {
+			// Socket.IO: 42<ackId>["method", data]
+			$payload = '42' . $ackId . json_encode(array($method, $data), JSON_UNESCAPED_SLASHES);
+			self::sendRaw($socketKey, $payload);
+		} else {
+			// Bare: {"event":"method", "data":..., "ack": ackId}
+			self::send($socketKey, array('event' => $method, 'data' => $data, 'ack' => $ackId));
+		}
+	}
+
+	/**
+	 * Route an ack response from a client back to the child that
+	 * initiated the RPC call.
+	 * @return boolean True if this was an RPC ack and was handled
+	 */
+	static function handleRpcResponse($ackId, $result)
+	{
+		if (!isset(self::$pendingRpc[$ackId])) return false;
+
+		$pending = self::$pendingRpc[$ackId];
+		unset(self::$pendingRpc[$ackId]);
+
+		// Send response back to child via IPC pipe
+		$response = json_encode(array(
+			'_rpc' => $pending['rpcId'],
+			'result' => $result,
+		), JSON_UNESCAPED_SLASHES);
+		$packet = pack('N', strlen($response)) . $response;
+		@fwrite($pending['pipe'], $packet);
+		return true;
+	}
 }

src/Q/socket.js

@@ -0,0 +1,102 @@
+/**
+ * QSocket — minimal WebSocket client for Qbix Server.
+ * ~80 lines. No dependencies. Plain JSON over bare WebSocket.
+ *
+ *   var socket = new QSocket('/ws');
+ *   socket.on('chat/message', function(data) { ... });
+ *   socket.emit('chat/message', {text: 'hi'}, function(res) { ... });
+ */
+(function(root) {
+  'use strict';
+  function QSocket(path, opts) {
+    opts = opts || {};
+    var self = this;
+    self._l = {};
+    self._h = {};
+    self._a = {};
+    self._n = 0;
+    self._q = [];
+    self._rc = opts.reconnect !== false;
+    self._d = opts.delay || 1000;
+    self._md = opts.maxDelay || 30000;
+    self._cd = self._d;
+
+    var loc = root.location || {};
+    var proto = (loc.protocol === 'https:') ? 'wss://' : 'ws://';
+    self._url = (path.indexOf('ws') === 0) ? path : proto + loc.host + path;
+
+    self._open = function() {
+      var ws = self.ws = new WebSocket(self._url);
+      ws.onopen = function() {
+        self._cd = self._d;
+        while (self._q.length) ws.send(self._q.shift());
+        self._emit('connect');
+      };
+      ws.onmessage = function(e) {
+        var m; try { m = JSON.parse(e.data); } catch(x) { return; }
+        // Ack response
+        if (m.ack != null && !m.event && self._a[m.ack]) {
+          self._a[m.ack](m.data); delete self._a[m.ack]; return;
+        }
+        // Server RPC call (event + ack, no pending callback)
+        if (m.event && m.ack != null && !self._a[m.ack]) {
+          var h = self._h[m.event], r;
+          try { r = h ? h(m.data) : null; } catch(x) { r = null; }
+          if (r && typeof r.then === 'function') {
+            r.then(function(v) { self._ack(m.ack, v); })
+             ['catch'](function() { self._ack(m.ack, null); });
+          } else {
+            self._ack(m.ack, r);
+          }
+          return;
+        }
+        if (m.event) self._emit(m.event, m.data);
+      };
+      ws.onclose = function() {
+        self._emit('disconnect');
+        if (self._rc) {
+          self._cd = Math.min(self._cd * 1.5, self._md);
+          setTimeout(self._open, self._cd);
+        }
+      };
+      ws.onerror = function() { ws.close(); };
+    };
+
+    self._emit = function(ev, d) {
+      var ls = self._l[ev]; if (!ls) return;
+      for (var i = 0; i < ls.length; i++) ls[i](d);
+    };
+    self._ack = function(id, d) {
+      self._send(JSON.stringify({ack: id, data: d}));
+    };
+    self._send = function(s) {
+      if (self.ws && self.ws.readyState === 1) self.ws.send(s);
+      else self._q.push(s);
+    };
+    self._open();
+  }
+
+  QSocket.prototype.on = function(ev, fn) {
+    if (!this._l[ev]) this._l[ev] = [];
+    this._l[ev].push(fn); return this;
+  };
+  QSocket.prototype.off = function(ev, fn) {
+    if (!fn) { delete this._l[ev]; return this; }
+    if (this._l[ev]) this._l[ev] = this._l[ev].filter(function(f) { return f !== fn; });
+    return this;
+  };
+  QSocket.prototype.emit = function(ev, data, cb) {
+    var m = {event: ev, data: data != null ? data : null};
+    if (typeof cb === 'function') { m.ack = ++this._n; this._a[m.ack] = cb; }
+    this._send(JSON.stringify(m)); return this;
+  };
+  QSocket.prototype.handle = function(method, fn) {
+    this._h[method] = fn; return this;
+  };
+  QSocket.prototype.close = function() {
+    this._rc = false; if (this.ws) this.ws.close();
+  };
+
+  if (typeof module !== 'undefined' && module.exports) module.exports = QSocket;
+  else root.QSocket = QSocket;
+})(typeof window !== 'undefined' ? window : this);