diff --git a/README.md b/README.md index 6c988d8..a3fb116 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,28 @@ This is in the early stages of development. Don't use this for anything importan ## Udo's C++ Entry Points -The aim of this project is to make a PHP-like runtime that enables server-side "scripting" using C/C++. At the core is a multi-worker FastCGI server that can be talked to from Nginx or similar front-end servers. UCE has a shared-nothing isolated architecture to serve page requests. To minimize the potential for memory leaks, UCE uses a per-request memory arena. UCE also provides a PHP-like API to ease web development. Advanced features such as an integrated WebSockets broker are planned. UCE aims to use minimal dependencies (at the moment, the only dependency is the Clang compiler). UCE pages are automatically recompiled and dynamically reloaded as necessary. +The aim of this project is to make a PHP-like runtime that enables server-side "scripting" using C/C++. At the core is a multi-worker FastCGI server that can be talked to from Nginx or similar front-end servers. UCE has a shared-nothing isolated architecture to serve page requests. To minimize the potential for memory leaks, UCE uses a per-request memory arena. UCE also provides a PHP-like API to ease web development. UCE aims to use minimal dependencies (at the moment, the only dependency is the Clang compiler). UCE pages are automatically recompiled and dynamically reloaded as necessary. + +## WebSockets + +WebSocket-enabled pages can now expose both: + +- `RENDER() { }` for normal HTTP rendering +- `WS() { }` for inbound WebSocket messages on the same `.ws.uce` page + +The runtime keeps the socket lifecycle in-process and exposes a low-boilerplate API to page code: + +- `ws_message()` +- `ws_connection_id()` +- `ws_scope()` +- `ws_connections([scope])` +- `ws_connection_count([scope])` +- `ws_send(message[, scope])` +- `ws_broadcast(message[, scope])` +- `ws_send_to(connection_id, message)` +- `ws_close([connection_id])` + +By default, the WebSocket scope is the current page file, so `ws_send()` broadcasts to other clients connected to that same `.ws.uce` endpoint. ## Service Setup @@ -26,6 +47,8 @@ scripts/systemd/manage-uce-service.sh logs 200 The service runs the runtime from the repository root so `COMPILER_SYS_PATH` resolves correctly and nginx can forward `.uce` requests to the Unix socket defined in `/etc/uce/settings.cfg`. +For deployed WebSockets, nginx should proxy `.ws.uce` requests to the runtime's built-in HTTP listener instead of `fastcgi_pass`. The current OpenFU deployment already does this and keeps HTTP/WebSocket ownership on a single worker so page-scoped broadcasts stay coherent. + # API Memcache Functions diff --git a/src/fastcgi/src/fcgicc.cc b/src/fastcgi/src/fcgicc.cc index 3df231e..aea2b63 100644 --- a/src/fastcgi/src/fcgicc.cc +++ b/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::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() { diff --git a/src/fastcgi/src/fcgicc.h b/src/fastcgi/src/fcgicc.h index b8eb867..f2999c6 100644 --- a/src/fastcgi/src/fcgicc.h +++ b/src/fastcgi/src/fcgicc.h @@ -48,6 +48,7 @@ public: std::function on_request = 0; std::function on_data = 0; std::function on_complete = 0; + std::function 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 server_socket_types; std::map 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); diff --git a/src/lib/compiler.cpp b/src/lib/compiler.cpp index df7de95..0970d41 100644 --- a/src/lib/compiler.cpp +++ b/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); } - diff --git a/src/lib/compiler.h b/src/lib/compiler.h index 52e4e7f..152badf 100644 --- a/src/lib/compiler.h +++ b/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); diff --git a/src/lib/sys.cpp b/src/lib/sys.cpp index b611111..75abde6 100644 --- a/src/lib/sys.cpp +++ b/src/lib/sys.cpp @@ -431,7 +431,7 @@ std::map workers; #include #include -void spawn_subprocess(std::function exec_after_spawn) +pid_t spawn_subprocess(std::function exec_after_spawn) { parent_pid = getpid(); pid_t p; @@ -442,6 +442,7 @@ void spawn_subprocess(std::function 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 exec_after_spawn) w.pid = p; workers[w.pid] = w; printf("(P) child procress spawned: PID %i\n", p); + return(p); } } diff --git a/src/lib/sys.h b/src/lib/sys.h index 28776d0..1e6fcb5 100644 --- a/src/lib/sys.h +++ b/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); diff --git a/src/lib/types.cpp b/src/lib/types.cpp index e442f37..6d8a771 100644 --- a/src/lib/types.cpp +++ b/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); } - diff --git a/src/lib/types.h b/src/lib/types.h index c34ab46..6464aec 100644 --- a/src/lib/types.h +++ b/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 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; diff --git a/src/lib/uri.cpp b/src/lib/uri.cpp index f5737e6..03cc2fb 100644 --- a/src/lib/uri.cpp +++ b/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); +} diff --git a/src/lib/uri.h b/src/lib/uri.h index 58fbe83..7f758bb 100644 --- a/src/lib/uri.h +++ b/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(const_cast(buffer + offset)); - offset += 4; - } - - // Parse the frame payload - payload = reinterpret_cast(const_cast(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); }; diff --git a/src/linux_fastcgi.cpp b/src/linux_fastcgi.cpp index 6631f3d..384356d 100644 --- a/src/linux_fastcgi.cpp +++ b/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(&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); } diff --git a/test/websockets.ws.uce b/test/websockets.ws.uce index f38b4b9..96d0a95 100644 --- a/test/websockets.ws.uce +++ b/test/websockets.ws.uce @@ -1,42 +1,197 @@ +String clean_chat_field(String raw, u32 max_length) +{ + String value = trim(raw); + value = replace(value, "\r", " "); + value = replace(value, "\n", " "); + if(value.length() > max_length) + value.resize(max_length); + return(value); +} + +DTree chat_event(String type, String body) +{ + DTree event; + event["type"] = type; + event["body"] = body; + event["connection_id"] = ws_connection_id(); + event["scope"] = first(context->params["DOCUMENT_URI"], ws_scope()); + event["online"] = (f64)ws_connection_count(); + event["at"] = gmdate("%H:%M:%S"); + return(event); +} RENDER() { - String raw = first(context->post["raw"], "today"); - String server_url = context->params["HTTP_HOST"]+context->params["DOCUMENT_URI"]; + String ws_url = context->params["DOCUMENT_URI"]; + u64 online = ws_connection_count(); <> - -

- UCE Test: - WebSocket Test -

-
-
- - + + +
+

+ UCE Test: + WebSocket Chat +

+
+
Page scope:
+
Connected right now:
+
Connecting...
-
- +
+
- - 

+			

+				

+					
+					
+				

+				

+					
+				

+			

+		

 		
-		
params) ?>

 	
 }
+
+WS()
+{
+	DTree payload = json_decode(ws_message());
+	String type = clean_chat_field(payload["type"].to_string(), 24);
+	String name = clean_chat_field(payload["name"].to_string(), 32);
+	String body = clean_chat_field(payload["body"].to_string(), 500);
+
+	if(name == "")
+		name = "guest";
+
+	if(type == "join")
+	{
+		ws_send(json_encode(chat_event("join", name + " joined the room")));
+		return;
+	}
+
+	if(type == "message" && body != "")
+	{
+		DTree event = chat_event("message", body);
+		event["name"] = name;
+		ws_send(json_encode(event));
+		return;
+	}
+
+	if(type != "")
+	{
+		ws_send_to(ws_connection_id(), json_encode(chat_event("notice", "Unknown message type: " + type)));
+	}
+}