Add WebSocket support and enhance chat functionality

- Implement WebSocket handling in the FastCGI server, allowing for real-time communication.
- Introduce functions for managing WebSocket connections, broadcasting messages, and sending to specific connections.
- Create a chat interface in the `websockets.ws.uce` file, including message handling and user notifications.
- Refactor existing code to accommodate WebSocket integration, ensuring compatibility with HTTP requests.
- Update documentation to reflect new features and usage instructions for WebSocket functionality.

src/fastcgi/src/fcgicc.cc

@@ -47,6 +47,19 @@
 
 #include "../fastcgi_devkit/fastcgi.h"
 
+static String
+make_http_text_response(String status_line, String body, String extra_headers = "")
+{
+	return(
+		status_line + "\r\n"
+		"Content-Type: text/plain; charset=utf-8\r\n" +
+		extra_headers +
+		"Content-Length: " + std::to_string(body.length()) + "\r\n"
+		"Connection: close\r\n\r\n" +
+		body
+	);
+}
+
 void
 FastCGIServer::shutdown()
 {
@@ -187,6 +200,23 @@ FastCGIServer::send_output_buffer(Connection& con)
 	return write_result;
 }
 
+void
+FastCGIServer::close_http_listeners()
+{
+	for(std::vector<int>::iterator it = server_sockets.begin(); it != server_sockets.end();)
+	{
+		int socket_handle = *it;
+		if(server_socket_types[socket_handle] == 'H')
+		{
+			close(socket_handle);
+			server_socket_types.erase(socket_handle);
+			it = server_sockets.erase(it);
+			continue;
+		}
+		++it;
+	}
+}
+
 void
 FastCGIServer::process(int timeout_ms)
 {
@@ -257,7 +287,11 @@ FastCGIServer::process(int timeout_ms)
 					throw std::runtime_error("read() on socket failed");
 				if (read_result == 0)
 				{
-					if(it->second->type == 'H' && it->second->input_buffer != "")
+					if(it->second->type == 'H' && it->second->is_websocket)
+					{
+						it->second->close_socket = true;
+					}
+					else if(it->second->type == 'H' && it->second->input_buffer != "")
 					{
 						process_http_request(
 							*client_sockets[it->second->client_socket]->requests[it->second->client_socket],
@@ -277,11 +311,18 @@ FastCGIServer::process(int timeout_ms)
 				it->second->input_buffer.append(buffer, read_result);
 					if(it->second->type == 'H')
 					{
-						process_http_request(
-							*client_sockets[it->second->client_socket]->requests[it->second->client_socket],
-							it->second->input_buffer);
-						if(it->second->close_socket || !it->second->output_buffer.empty())
-							it->second->input_buffer = "";
+						if(it->second->is_websocket)
+						{
+							process_websocket_input(*it->second);
+						}
+						else
+						{
+							process_http_request(
+								*client_sockets[it->second->client_socket]->requests[it->second->client_socket],
+								it->second->input_buffer);
+							if(it->second->close_socket || !it->second->output_buffer.empty())
+								it->second->input_buffer = "";
+						}
 					}
 					else
 					{
@@ -293,7 +334,8 @@ FastCGIServer::process(int timeout_ms)
 		if (!it->second->output_buffer.empty() &&
 				FD_ISSET(read_socket, &fs_write))
 		{
-			write_fgci(*it->second);
+			if(it->second->type == 'F')
+				write_fgci(*it->second);
 			send_output_buffer(*it->second);
 		}
 
@@ -341,16 +383,68 @@ FastCGIServer::process_http_request(FastCGIRequest& request, String& data)
 		}
 	}
 
-	request.in = data.substr(header_end + 4);
 	u64 content_length = int_val(first(request.params["CONTENT_LENGTH"], "0"));
-	if(request.in.length() < content_length)
+	u64 request_size = header_end + 4 + content_length;
+	if(data.length() < request_size)
 		return;
+	request.in = data.substr(header_end + 4, content_length);
 	request.flags.input_closed = true;
+	data.erase(0, request_size);
 
-	if(request.params["HTTP_UPGRADE"] == "websocket")
+	if(to_lower(request.params["HTTP_UPGRADE"]) == "websocket")
 	{
-		printf("(i) got upgrade request %s\n", var_dump(request.params).c_str());
-		client_sockets[request.resources.client_socket]->close_socket = true;
+		Connection* connection = client_sockets[request.resources.client_socket];
+		String websocket_connection = to_lower(request.params["HTTP_CONNECTION"]);
+		String websocket_key = trim(request.params["HTTP_SEC_WEBSOCKET_KEY"]);
+		String websocket_version = trim(request.params["HTTP_SEC_WEBSOCKET_VERSION"]);
+		if(websocket_connection.find("upgrade") == String::npos)
+		{
+			connection->output_buffer += make_http_text_response(
+				"HTTP/1.1 400 Bad Request",
+				"missing Connection: Upgrade header"
+			);
+			connection->close_socket = true;
+			return;
+		}
+		if(websocket_key == "")
+		{
+			connection->output_buffer += make_http_text_response(
+				"HTTP/1.1 400 Bad Request",
+				"missing Sec-WebSocket-Key header"
+			);
+			connection->close_socket = true;
+			return;
+		}
+		if(websocket_version != "" && websocket_version != "13")
+		{
+			connection->output_buffer += make_http_text_response(
+				"HTTP/1.1 426 Upgrade Required",
+				"unsupported websocket version",
+				"Sec-WebSocket-Version: 13\r\n"
+			);
+			connection->close_socket = true;
+			return;
+		}
+
+		connection->is_websocket = true;
+		connection->websocket_connection_id = std::to_string(getpid()) + ":" + std::to_string(connection->client_socket);
+		connection->websocket_scope = first(
+			request.params["SCRIPT_FILENAME"],
+			request.params["DOCUMENT_URI"],
+			request.params["REQUEST_URI"]
+		);
+		request.resources.is_websocket = true;
+		request.resources.websocket_connection_id = connection->websocket_connection_id;
+		request.resources.websocket_scope = connection->websocket_scope;
+
+		connection->output_buffer +=
+			"HTTP/1.1 101 Switching Protocols\r\n"
+			"Upgrade: websocket\r\n"
+			"Connection: Upgrade\r\n"
+			"Sec-WebSocket-Accept: " + ws_make_accept_key(websocket_key) + "\r\n\r\n";
+
+		if(!data.empty())
+			process_websocket_input(*connection);
 	}
 	else
 	{
@@ -367,6 +461,134 @@ FastCGIServer::process_http_request(FastCGIRequest& request, String& data)
 	}
 }
 
+void
+FastCGIServer::process_websocket_input(Connection& connection)
+{
+	while(!connection.input_buffer.empty())
+	{
+		WSFrame frame;
+		String error;
+		if(!frame.parse(connection.input_buffer, error))
+		{
+			if(error != "")
+			{
+				connection.output_buffer += ws_close_frame(1002, error);
+				connection.close_socket = true;
+			}
+			return;
+		}
+
+		connection.input_buffer.erase(0, frame.frame_length);
+
+		if(!frame.mask_bit)
+		{
+			connection.output_buffer += ws_close_frame(1002, "client frames must be masked");
+			connection.close_socket = true;
+			return;
+		}
+
+		if(!frame.is_final_fragment || frame.opcode == 0x0)
+		{
+			connection.output_buffer += ws_close_frame(1003, "fragmented frames are not supported");
+			connection.close_socket = true;
+			return;
+		}
+
+		switch(frame.opcode)
+		{
+			case 0x1:
+			{
+				RequestList::iterator it = connection.requests.find(connection.client_socket);
+				if(it != connection.requests.end() && on_websocket_message)
+				{
+					it->second->resources.is_websocket = true;
+					it->second->resources.websocket_connection_id = connection.websocket_connection_id;
+					it->second->resources.websocket_scope = connection.websocket_scope;
+					on_websocket_message(*it->second, frame.payload);
+				}
+				break;
+			}
+			case 0x8:
+				connection.output_buffer += ws_close_frame();
+				connection.close_socket = true;
+				return;
+			case 0x9:
+				connection.output_buffer += ws_encode_frame(frame.payload, 0xA);
+				break;
+			case 0xA:
+				break;
+			default:
+				connection.output_buffer += ws_close_frame(1003, "unsupported websocket opcode");
+				connection.close_socket = true;
+				return;
+		}
+	}
+}
+
+bool
+FastCGIServer::websocket_send_to(String connection_id, String message)
+{
+	for(auto& item : client_sockets)
+	{
+		Connection* connection = item.second;
+		if(connection->is_websocket && connection->websocket_connection_id == connection_id)
+		{
+			connection->output_buffer += ws_encode_frame(message);
+			return(true);
+		}
+	}
+	return(false);
+}
+
+u64
+FastCGIServer::websocket_broadcast(String scope, String message)
+{
+	u64 sent = 0;
+	for(auto& item : client_sockets)
+	{
+		Connection* connection = item.second;
+		if(!connection->is_websocket)
+			continue;
+		if(scope != "" && connection->websocket_scope != scope)
+			continue;
+		connection->output_buffer += ws_encode_frame(message);
+		sent += 1;
+	}
+	return(sent);
+}
+
+StringList
+FastCGIServer::websocket_connection_ids(String scope)
+{
+	StringList result;
+	for(auto& item : client_sockets)
+	{
+		Connection* connection = item.second;
+		if(!connection->is_websocket)
+			continue;
+		if(scope != "" && connection->websocket_scope != scope)
+			continue;
+		result.push_back(connection->websocket_connection_id);
+	}
+	return(result);
+}
+
+bool
+FastCGIServer::websocket_close(String connection_id, u16 status_code, String reason)
+{
+	for(auto& item : client_sockets)
+	{
+		Connection* connection = item.second;
+		if(connection->is_websocket && connection->websocket_connection_id == connection_id)
+		{
+			connection->output_buffer += ws_close_frame(status_code, reason);
+			connection->close_socket = true;
+			return(true);
+		}
+	}
+	return(false);
+}
+
 void
 FastCGIServer::process_forever()
 {

src/fastcgi/src/fcgicc.h

@@ -48,6 +48,7 @@ public:
 	std::function<int(FastCGIRequest&)> on_request = 0;
 	std::function<int(FastCGIRequest&)> on_data = 0;
 	std::function<int(FastCGIRequest&)> on_complete = 0;
+	std::function<int(FastCGIRequest&, const String&)> on_websocket_message = 0;
 
 	int listen(unsigned tcp_port);
 	int listen_http(unsigned tcp_port);
@@ -68,6 +69,9 @@ public:
 		std::string output_buffer;
 		bool close_responsibility = false;
 		bool close_socket = false;
+		bool is_websocket = false;
+		String websocket_connection_id;
+		String websocket_scope;
 		char type = 'F'; // F = FastCGI, H = HttpServer
 	};
 
@@ -79,7 +83,13 @@ public:
 	std::map<int, char> server_socket_types;
 	std::map<int, Connection*> client_sockets;
 
+	void close_http_listeners();
 	void read_fgci(Connection&);
+	void process_websocket_input(Connection&);
+	bool websocket_send_to(String connection_id, String message);
+	u64 websocket_broadcast(String scope, String message);
+	StringList websocket_connection_ids(String scope = "");
+	bool websocket_close(String connection_id, u16 status_code = 1000, String reason = "");
 	static void request_write_fgci(Connection&, RequestID, FastCGIRequest&);
 	static void write_fgci(Connection&);
 	static Pairs parse_pairs_fcgi(const char*, std::string::size_type);

src/lib/compiler.cpp

@@ -202,6 +202,7 @@ void load_shared_unit(Request* context, SharedUnit* su, String file_name)
 	//setup_unit_paths(context, su, file_name);
 
 	su->on_render = 0;
+	su->on_websocket = 0;
 	su->on_setup = 0;
 	su->compiler_messages = "";
 
@@ -223,6 +224,9 @@ void load_shared_unit(Request* context, SharedUnit* su, String file_name)
 		if ((error = dlerror()) != NULL)
 			printf("Error - %s in %s\n", error, su->file_name.c_str());
 		su->on_render = (call_handler)dlsym(su->so_handle, "render");
+		dlerror();
+		su->on_websocket = (call_handler)dlsym(su->so_handle, "websocket");
+		dlerror();
 		su->api_declarations = split(file_get_contents(su->api_file_name), "\n");
 		//else
 		//	printf("(i) loaded unit %s\n", su->file_name.c_str());
@@ -403,6 +407,31 @@ void compiler_invoke(Request* context, String file_name, DTree& call_param)
 	}
 }
 
+void compiler_invoke_websocket(Request* context, String file_name, DTree& call_param)
+{
+	auto su = compiler_load_shared_unit(context, file_name, "", false);
+	if(!su)
+		return;
+
+	if(!su->on_setup)
+	{
+		printf("internal error: set_current_request() not defined in %s\n", file_name.c_str());
+		return;
+	}
+
+	if(!su->on_websocket)
+	{
+		printf("no WS() entry point in %s\n", file_name.c_str());
+		return;
+	}
+
+	String prev_wd = get_cwd();
+	set_cwd(su->src_path);
+	su->on_setup(context);
+	su->on_websocket(call_param);
+	set_cwd(prev_wd);
+}
+
 void render_file(String file_name)
 {
 	//printf("(i) render_file(%s)\n", file_name.c_str());
@@ -461,4 +490,3 @@ DTree* call_file(String file_name, String function_name, DTree* call_param)
 	}
 	return(result);
 }
-

src/lib/compiler.h

@@ -1,4 +1,5 @@
 #define RENDER() extern "C" void render(DTree& call)
+#define WS() extern "C" void websocket(DTree& call)
 #define EXPORT extern "C"
 
 String process_html_literal(Request* context, SharedUnit* su, String content);
@@ -8,6 +9,7 @@ void load_shared_unit(Request* context, SharedUnit* su, String file_name);
 void compile_shared_unit(Request* context, SharedUnit* su, String file_name);
 SharedUnit* get_shared_unit(Request* context, String file_name, bool opt_so_optional = false);
 void compiler_invoke(Request* context, String file_name, DTree& call_param);
+void compiler_invoke_websocket(Request* context, String file_name, DTree& call_param);
 SharedUnit* compiler_load_shared_unit(Request* context, String file_name, String current_path = "", bool opt_so_optional = false);
 
 SharedUnit* load_file(String file_name);

src/lib/sys.cpp

@@ -431,7 +431,7 @@ std::map<pid_t, Worker> workers;
 #include <sys/resource.h>
 #include <sys/prctl.h>
 
-void spawn_subprocess(std::function<void()> exec_after_spawn)
+pid_t spawn_subprocess(std::function<void()> exec_after_spawn)
 {
 	parent_pid = getpid();
 	pid_t p;
@@ -442,6 +442,7 @@ void spawn_subprocess(std::function<void()> exec_after_spawn)
 		//printf("(C) child procress started, PID:%i\n", my_pid);
 		prctl(PR_SET_PDEATHSIG, SIGHUP);
 		exec_after_spawn();
+		return(0);
 	}
 	else
 	{
@@ -449,6 +450,7 @@ void spawn_subprocess(std::function<void()> exec_after_spawn)
 		w.pid = p;
 		workers[w.pid] = w;
 		printf("(P) child procress spawned: PID %i\n", p);
+		return(p);
 	}
 }
 

src/lib/sys.h

@@ -36,6 +36,16 @@ void socket_close(u64 sockfd);
 bool socket_write(u64 sockfd, String data);
 String socket_read(u64 sockfd, u32 max_length = 1024*128, u32 timeout = 1);
 
+String ws_message();
+String ws_connection_id();
+String ws_scope();
+StringList ws_connections(String scope = "");
+u64 ws_connection_count(String scope = "");
+bool ws_send(String message, String scope = "");
+u64 ws_broadcast(String message, String scope = "");
+bool ws_send_to(String connection_id, String message);
+bool ws_close(String connection_id = "");
+
 String memcache_escape_key(String key);
 StringList memcache_escape_keys(StringList keys);
 u64 memcache_connect(String host = "127.0.0.1", short port = 11211);

src/lib/types.cpp

@@ -47,7 +47,10 @@ void Request::ob_start()
 
 Request::~Request()
 {
+	for(auto* stream : ob_stack)
+		delete stream;
+	ob_stack.clear();
+	ob = 0;
 	for(auto& sockfd : resources.sockets)
 		close(sockfd);
 }
-

src/lib/types.h

@@ -78,6 +78,7 @@ struct SharedUnit {
 
 	request_handler on_setup;
 	call_handler on_render;
+	call_handler on_websocket;
 
 	String compiler_messages;
 	time_t last_compiled;
@@ -169,6 +170,9 @@ struct Request {
 		std::vector<void*> mysql_connections;
 		u64 client_socket = 0;
 		u64 server_socket = 0;
+		bool is_websocket = false;
+		String websocket_connection_id = "";
+		String websocket_scope = "";
 		std::string params_buffer;
 	} resources;
 

src/lib/uri.cpp

@@ -1,5 +1,32 @@
 #include "uri.h"
 
+static String base64_encode(String raw)
+{
+	static const char* chars =
+		"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+		"abcdefghijklmnopqrstuvwxyz"
+		"0123456789+/";
+
+	String result;
+	int val = 0;
+	int bits = -6;
+	for(unsigned char c : raw)
+	{
+		val = (val << 8) + c;
+		bits += 8;
+		while(bits >= 0)
+		{
+			result.append(1, chars[(val >> bits) & 0x3F]);
+			bits -= 6;
+		}
+	}
+	if(bits > -6)
+		result.append(1, chars[((val << 8) >> (bits + 8)) & 0x3F]);
+	while(result.length() % 4 != 0)
+		result.append(1, '=');
+	return(result);
+}
+
 String var_dump(URI uri, String prefix, String postfix)
 {
 	return(
@@ -427,3 +454,106 @@ void session_destroy(String session_name)
 		context->session_id = "";
 	}
 }
+
+String ws_make_accept_key(String client_key)
+{
+	return(base64_encode(gen_sha1(
+		client_key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11",
+		true
+	)));
+}
+
+String ws_encode_frame(String payload, u8 opcode, bool is_final_fragment)
+{
+	String frame;
+	u8 first_byte = opcode & 0x0F;
+	if(is_final_fragment)
+		first_byte |= 0x80;
+	frame.append(1, first_byte);
+
+	u64 payload_length = payload.length();
+	if(payload_length <= 125)
+	{
+		frame.append(1, (char)payload_length);
+	}
+	else if(payload_length <= 0xFFFF)
+	{
+		frame.append(1, 126);
+		frame.append(1, (char)((payload_length >> 8) & 0xFF));
+		frame.append(1, (char)(payload_length & 0xFF));
+	}
+	else
+	{
+		frame.append(1, 127);
+		for(int shift = 56; shift >= 0; shift -= 8)
+			frame.append(1, (char)((payload_length >> shift) & 0xFF));
+	}
+
+	frame += payload;
+	return(frame);
+}
+
+String ws_close_frame(u16 status_code, String reason)
+{
+	String payload;
+	payload.append(1, (char)((status_code >> 8) & 0xFF));
+	payload.append(1, (char)(status_code & 0xFF));
+	payload += reason;
+	return(ws_encode_frame(payload, 0x8));
+}
+
+bool WSFrame::parse(const String& buffer, String& error)
+{
+	error = "";
+	payload = "";
+
+	if(buffer.length() < 2)
+		return(false);
+
+	const unsigned char* raw = (const unsigned char*)buffer.data();
+	opcode = raw[0] & 0x0F;
+	is_final_fragment = (raw[0] & 0x80) != 0;
+	mask_bit = (raw[1] & 0x80) != 0;
+	payload_length = raw[1] & 0x7F;
+	header_length = 2;
+
+	if(payload_length == 126)
+	{
+		if(buffer.length() < 4)
+			return(false);
+		payload_length = ((u64)raw[2] << 8) | (u64)raw[3];
+		header_length = 4;
+	}
+	else if(payload_length == 127)
+	{
+		if(buffer.length() < 10)
+			return(false);
+		payload_length = 0;
+		for(u32 i = 0; i < 8; i++)
+			payload_length = (payload_length << 8) | (u64)raw[2 + i];
+		header_length = 10;
+	}
+
+	u64 mask_offset = header_length;
+	if(mask_bit)
+		header_length += 4;
+
+	frame_length = header_length + payload_length;
+	if(frame_length < header_length)
+	{
+		error = "invalid websocket frame length";
+		return(false);
+	}
+	if(buffer.length() < frame_length)
+		return(false);
+
+	payload.assign(buffer.data() + header_length, payload_length);
+	if(mask_bit)
+	{
+		const unsigned char* mask = raw + mask_offset;
+		for(u64 i = 0; i < payload.length(); i++)
+			payload[i] = payload[i] ^ mask[i % 4];
+	}
+
+	return(true);
+}

src/lib/uri.h

@@ -17,69 +17,20 @@ StringMap load_session_data(String session_id);
 void save_session_data(String session_id, StringMap data);
 String session_start(String session_name = "uce-session");
 void session_destroy(String session_name = "uce-session");
+String ws_make_accept_key(String client_key);
+String ws_encode_frame(String payload, u8 opcode = 0x1, bool is_final_fragment = true);
+String ws_close_frame(u16 status_code = 1000, String reason = "");
 
 struct WSFrame {
 
-	u8 opcode;
-	u8 is_final_fragment;
-	u8 mask_bit;
-	u64 payload_length;
-	u64 offset;
-	u8* payload;
+	u8 opcode = 0;
+	bool is_final_fragment = false;
+	bool mask_bit = false;
+	u64 payload_length = 0;
+	u64 header_length = 0;
+	u64 frame_length = 0;
+	String payload;
 
-	WSFrame(String& bufstr)
-	{
-		const char* buffer = bufstr.c_str();
-		u64 buffer_size = bufstr.size();
-		opcode = buffer[0] & 0x0F;
-		is_final_fragment = buffer[0] & 0x80;
-		mask_bit = buffer[1] & 0x80;
-		payload_length = buffer[1] & 0x7F;
-
-		offset = 2;
-		if (payload_length == 126) {
-			payload_length = ((u64) buffer[2] << 8) | buffer[3];
-			offset = 4;
-		} else if (payload_length == 127) {
-			payload_length = ((u64) buffer[2] << 56) |
-				((u64) buffer[3] << 48) |
-				((u64) buffer[4] << 40) |
-				((u64) buffer[5] << 32) |
-				((u64) buffer[6] << 24) |
-				((u64) buffer[7] << 16) |
-				((u64) buffer[8] << 8) |
-				buffer[9];
-			offset = 10;
-		}
-
-		uint8_t* maskingKey = nullptr;
-		if (mask_bit) {
-			maskingKey = reinterpret_cast<uint8_t*>(const_cast<char*>(buffer + offset));
-			offset += 4;
-		}
-
-		// Parse the frame payload
-		payload = reinterpret_cast<uint8_t*>(const_cast<char*>(buffer + offset));
-		if (mask_bit) {
-			for (int i = 0; i < payload_length; i++) {
-				payload[i] = payload[i] ^ maskingKey[i % 4];
-			}
-		}
-
-		// Print the parsed frame
-		std::cout << "Parsed WebSocket frame:" << std::endl;
-		std::cout << "  Opcode: " << (int) opcode << std::endl;
-		std::cout << "  Is final fragment: " << (is_final_fragment ? "true" : "false") << std::endl;
-		std::cout << "  Payload length: " << payload_length << std::endl;
-		if (mask_bit) {
-			std::cout << "  Masking key: ";
-			for (int i = 0; i < 4; i++) {
-				std::cout << std::hex << (int) maskingKey[i] << " ";
-			}
-			std::cout << std::endl;
-		}
-		std::cout << "  Payload: " << std::endl << std::string((const char*) payload, payload_length) << std::endl;
-
-	}
+	bool parse(const String& buffer, String& error);
 
 };

src/linux_fastcgi.cpp

@@ -5,6 +5,92 @@ ServerState server_state;
 #include "fastcgi/src/fcgicc.cc"
 
 FastCGIServer server;
+pid_t http_worker_pid = 0;
+bool worker_accepts_http = false;
+
+Request* set_active_request(Request& request)
+{
+	Request* previous_context = context;
+	context = &request;
+	return(previous_context);
+}
+
+void restore_active_request(Request* previous_context)
+{
+	context = previous_context;
+}
+
+String current_ws_scope()
+{
+	if(!context)
+		return("");
+	return(first(
+		context->resources.websocket_scope,
+		context->params["SCRIPT_FILENAME"]
+	));
+}
+
+String normalize_ws_scope(String scope)
+{
+	if(scope == "")
+		return(current_ws_scope());
+	if(scope[0] == '/')
+		return(scope);
+	return(expand_path(scope, get_cwd()));
+}
+
+String ws_message()
+{
+	if(!context)
+		return("");
+	return(context->var["ws"]["message"].to_string());
+}
+
+String ws_connection_id()
+{
+	if(!context)
+		return("");
+	return(context->resources.websocket_connection_id);
+}
+
+String ws_scope()
+{
+	return(current_ws_scope());
+}
+
+StringList ws_connections(String scope)
+{
+	return(server.websocket_connection_ids(normalize_ws_scope(scope)));
+}
+
+u64 ws_connection_count(String scope)
+{
+	return(ws_connections(scope).size());
+}
+
+bool ws_send(String message, String scope)
+{
+	return(server.websocket_broadcast(normalize_ws_scope(scope), message) > 0);
+}
+
+u64 ws_broadcast(String message, String scope)
+{
+	return(server.websocket_broadcast(normalize_ws_scope(scope), message));
+}
+
+bool ws_send_to(String connection_id, String message)
+{
+	return(server.websocket_send_to(connection_id, message));
+}
+
+bool ws_close(String connection_id)
+{
+	if(connection_id == "")
+		connection_id = ws_connection_id();
+	if(connection_id == "")
+		return(false);
+	return(server.websocket_close(connection_id));
+}
 
 int handle_request(FastCGIRequest& request) {
     // This is always the first event to occur.  It occurs when the
@@ -39,7 +125,7 @@ int handle_complete(FastCGIRequest& request) {
     // both closed, and thus the request is complete.
 	// printf("(i) request handle\n");
 
-	context = &request;
+	Request* previous_context = set_active_request(request);
 	server_state.request_count += 1;
 	request.server = &server_state;
 	request.stats.time_start = microtime();
@@ -82,10 +168,60 @@ int handle_complete(FastCGIRequest& request) {
 		save_session_data(request.session_id, request.session);
 
 	cleanup_mysql_connections();
+	restore_active_request(previous_context);
 
     return 0;
 }
 
+int handle_websocket_message(FastCGIRequest& request, const String& message)
+{
+	Request event_request;
+	ByteStream ws_output;
+	DTree call_param;
+
+	Request* previous_context = set_active_request(event_request);
+	server_state.request_count += 1;
+	event_request.server = &server_state;
+	event_request.params = request.params;
+	event_request.params["REQUEST_METHOD"] = "WEBSOCKET";
+	event_request.get = parse_query(event_request.params["QUERY_STRING"]);
+	event_request.resources = request.resources;
+	event_request.stats.time_init = microtime();
+	event_request.stats.time_start = event_request.stats.time_init;
+	event_request.random_index = 0;
+	event_request.random_seed = gen_noise64(*reinterpret_cast<u64*>(&event_request.stats.time_start));
+	event_request.response_code = "WEBSOCKET";
+	event_request.header["Content-Type"] = context->server->config["CONTENT_TYPE"];
+	event_request.in = message;
+	event_request.ob = &ws_output;
+
+	if(event_request.params["HTTP_COOKIE"].length() > 0)
+		event_request.cookies = parse_cookies(event_request.params["HTTP_COOKIE"]);
+
+	event_request.var["ws"]["message"] = message;
+	event_request.var["ws"]["connection_id"] = request.resources.websocket_connection_id;
+	event_request.var["ws"]["scope"] = request.resources.websocket_scope;
+	event_request.var["ws"]["connection_count"] = (f64)server.websocket_connection_ids(request.resources.websocket_scope).size();
+	event_request.var["ws"]["document_uri"] = first(
+		request.params["DOCUMENT_URI"],
+		request.params["REQUEST_URI"]
+	);
+
+	call_param["message"] = message;
+	call_param["connection_id"] = request.resources.websocket_connection_id;
+	call_param["scope"] = request.resources.websocket_scope;
+	call_param["document_uri"] = event_request.var["ws"]["document_uri"].to_string();
+
+	compiler_invoke_websocket(&event_request, request.params["SCRIPT_FILENAME"], call_param);
+
+	if(event_request.session_id.length() > 0)
+		save_session_data(event_request.session_id, event_request.session);
+
+	cleanup_mysql_connections();
+	restore_active_request(previous_context);
+	return 0;
+}
+
 volatile bool termination_signal_received = false;
 
 void on_terminate(int sig)
@@ -114,9 +250,12 @@ void listen_for_connections()
 	}
 
 	signal(SIGSEGV, on_segfault);
+	if(!worker_accepts_http)
+		server.close_http_listeners();
 	server.on_request = &handle_request;
 	server.on_data = &handle_data;
 	server.on_complete = &handle_complete;
+	server.on_websocket_message = &handle_websocket_message;
 	for(;;)
 	{
 		server.process();
@@ -204,7 +343,12 @@ int main(int argc, char** argv)
 				}
 			}
 			if(!termination_signal_received)
-				spawn_subprocess(listen_for_connections);
+			{
+				worker_accepts_http = (http_worker_pid == 0 || workers.count(http_worker_pid) == 0);
+				pid_t child_pid = spawn_subprocess(listen_for_connections);
+				if(child_pid > 0 && worker_accepts_http)
+					http_worker_pid = child_pid;
+			}
 		}
 		sleep(1);
 	}