From 75224df0588073fc3117c141228ca063be176094 Mon Sep 17 00:00:00 2001 From: Onek8 Date: Sat, 3 Oct 2026 01:22:41 -0700 Subject: [PATCH] Static Webview - RTC - UDP - TCP - WSS --- README.md | 4 +- Sources/async_engine.h | 57 +--- Sources/connection_pool.cpp | 85 ++++- Sources/connection_pool.h | 4 +- Sources/httprequest.cpp | 162 ++++++++- Sources/httprequest.h | 5 +- Sources/lockfree_queue.h | 1 + Sources/main.cpp | 136 +++++++- Sources/rtc_bridge.cpp | 232 +++++++++++++ Sources/rtc_bridge.h | 54 +++ Sources/rtc_v8_bindings.cpp | 607 +++++++++++++++++++++++++++++++++ Sources/rtc_v8_bindings.h | 9 + Sources/socket_bridge.cpp | 232 ++++++++++++- Sources/socket_bridge.h | 7 + Sources/socket_v8_bindings.cpp | 147 ++++++++ Sources/socket_v8_bindings.h | 4 + Sources/websocket.cpp | 493 ++++++++++++++++---------- Sources/websocket.h | 12 +- kfile.js | 27 +- 19 files changed, 1994 insertions(+), 284 deletions(-) create mode 100644 Sources/rtc_bridge.cpp create mode 100644 Sources/rtc_bridge.h create mode 100644 Sources/rtc_v8_bindings.cpp create mode 100644 Sources/rtc_v8_bindings.h diff --git a/README.md b/README.md index 78e8198..03c8f50 100644 --- a/README.md +++ b/README.md @@ -40,14 +40,14 @@ cd apple-libressl-sdk git checkout fce8ffe30c9939fa1d71076745d9b3c811b8a90e ``` -**Webview for windodows** +**Webview for windows** ```bash # Windows needs the WebView2 SDK while mac and linux are OS implementations curl -L https://www.nuget.org/api/v2/package/Microsoft.Web.WebView2 -o webview2.zip powershell -Command "Expand-Archive -Path webview2.zip -DestinationPath webview2-tmp -Force" xcopy /E /I /Y webview2-tmp\build\native\include webview2\include\ mkdir webview2\lib\x64 -copy webview2-tmp\build\native\x64\WebView2Loader.dll.lib webview2\lib\x64\WebView2Loader.lib +copy webview2-tmp\build\native\x64\WebView2LoaderStatic.lib webview2\lib\x64\WebView2LoaderStatic.lib rmdir /S /Q webview2-tmp del webview2.zip ``` diff --git a/Sources/async_engine.h b/Sources/async_engine.h index 2ab7584..651f94d 100644 --- a/Sources/async_engine.h +++ b/Sources/async_engine.h @@ -1,5 +1,6 @@ #pragma once +#include "lockfree_queue.h" #include #include #include @@ -14,61 +15,7 @@ namespace EngineManager { // thread communication template -class LockFreeQueue { -private: - struct Node { - std::atomic data{nullptr}; - std::atomic next{nullptr}; - }; - - std::atomic head_{nullptr}; - std::atomic tail_{nullptr}; - -public: - LockFreeQueue() { - Node* dummy = new Node; - head_.store(dummy); - tail_.store(dummy); - } - - ~LockFreeQueue() { - while (Node* const old_head = head_.load()) { - head_.store(old_head->next); - delete old_head->data.load(); - delete old_head; - } - } - - void push(T item) { - Node* new_node = new Node; - T* data = new T(std::move(item)); - new_node->data.store(data); - - Node* prev_tail = tail_.exchange(new_node); - prev_tail->next.store(new_node); - } - - bool try_pop(T& result) { - Node* head = head_.load(); - Node* next = head->next.load(); - - if (next == nullptr) return false; - - T* data = next->data.load(); - if (data == nullptr) return false; - - result = *data; - delete data; - head_.store(next); - delete head; - return true; - } - - bool empty() const { - Node* head = head_.load(); - return (head->next.load() == nullptr); - } -}; +using LockFreeQueue = ::LockFreeQueue; // thread pool stealing class ThreadPool { diff --git a/Sources/connection_pool.cpp b/Sources/connection_pool.cpp index 3823950..b3b7078 100644 --- a/Sources/connection_pool.cpp +++ b/Sources/connection_pool.cpp @@ -7,6 +7,8 @@ #pragma comment(lib, "ws2_32.lib") #else #include +#include +#include #include #include #include @@ -34,7 +36,8 @@ ConnectionPool::~ConnectionPool() { } } -socket_t ConnectionPool::acquire_connection(const std::string& host, int port, bool is_ssl) { +socket_t ConnectionPool::acquire_connection(const std::string& host, int port, bool is_ssl, + std::atomic* cancel) { std::string key = make_key(host, port, is_ssl); { @@ -83,7 +86,65 @@ socket_t ConnectionPool::acquire_connection(const std::string& host, int port, b memcpy(&server_addr.sin_addr, host_entry->h_addr, host_entry->h_length); - if (::connect(sock, (struct sockaddr*)&server_addr, sizeof(server_addr)) == SOCKET_ERROR) { +#ifdef _WIN32 + u_long nb_mode = 1; + ioctlsocket(sock, FIONBIO, &nb_mode); +#else + int old_flags = fcntl(sock, F_GETFL, 0); + fcntl(sock, F_SETFL, old_flags | O_NONBLOCK); +#endif + + bool connected = (::connect(sock, (struct sockaddr*)&server_addr, sizeof(server_addr)) == 0); + if (!connected) { +#ifdef _WIN32 + int connect_err = WSAGetLastError(); + bool in_progress = (connect_err == WSAEWOULDBLOCK || connect_err == WSAEINPROGRESS); +#else + bool in_progress = (errno == EINPROGRESS); +#endif + auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(15000); + while (in_progress) { + if (cancel != nullptr && cancel->load()) { + in_progress = false; + break; + } + if (std::chrono::steady_clock::now() >= deadline) { + in_progress = false; + break; + } + fd_set wfds, efds; + FD_ZERO(&wfds); + FD_ZERO(&efds); + FD_SET(sock, &wfds); + FD_SET(sock, &efds); + struct timeval tv; + tv.tv_sec = 0; + tv.tv_usec = 50000; +#ifdef _WIN32 + int sel = ::select(0, nullptr, &wfds, &efds, &tv); +#else + int sel = ::select(static_cast(sock) + 1, nullptr, &wfds, &efds, &tv); +#endif + if (sel > 0) { + int so_err = 0; + socklen_t so_len = sizeof(so_err); + getsockopt(sock, SOL_SOCKET, SO_ERROR, (char*)&so_err, &so_len); + connected = (so_err == 0); + break; + } + if (sel < 0) break; + } + } + +#ifdef _WIN32 + nb_mode = 0; + ioctlsocket(sock, FIONBIO, &nb_mode); +#else + fcntl(sock, F_SETFL, old_flags); +#endif + + if (!connected) { kinc_log(KINC_LOG_LEVEL_ERROR, "Failed to connect to %s:%d", host.c_str(), port); #ifdef _WIN32 closesocket(sock); @@ -165,9 +226,23 @@ std::string ConnectionPool::make_key(const std::string& host, int port, bool is_ } bool ConnectionPool::is_socket_alive(socket_t socket) const { + fd_set rfds; + FD_ZERO(&rfds); + FD_SET(socket, &rfds); + struct timeval tv; + tv.tv_sec = 0; + tv.tv_usec = 0; +#ifdef _WIN32 + int sel = ::select(0, &rfds, nullptr, nullptr, &tv); +#else + int sel = ::select(static_cast(socket) + 1, &rfds, nullptr, nullptr, &tv); +#endif + if (sel < 0) return false; + if (sel == 0) return true; char test_byte; int result = ::recv(socket, &test_byte, 1, MSG_PEEK); - + if (result == 0) return false; + #ifdef _WIN32 if (result == SOCKET_ERROR) { int error = WSAGetLastError(); @@ -178,6 +253,6 @@ bool ConnectionPool::is_socket_alive(socket_t socket) const { return (errno == EWOULDBLOCK || errno == EAGAIN); } #endif - - return true; + + return false; } diff --git a/Sources/connection_pool.h b/Sources/connection_pool.h index 8054f6a..3ac407f 100644 --- a/Sources/connection_pool.h +++ b/Sources/connection_pool.h @@ -5,6 +5,7 @@ #include #include #include +#include #ifdef _WIN32 #include @@ -31,7 +32,8 @@ public: ConnectionPool() = default; ~ConnectionPool(); - socket_t acquire_connection(const std::string& host, int port, bool is_ssl = false); + socket_t acquire_connection(const std::string& host, int port, bool is_ssl = false, + std::atomic* cancel = nullptr); // returns connection to pool for reuse void release_connection(const std::string& host, int port, socket_t socket, bool is_ssl = false); diff --git a/Sources/httprequest.cpp b/Sources/httprequest.cpp index 1bd59bf..e0ca230 100644 --- a/Sources/httprequest.cpp +++ b/Sources/httprequest.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #ifdef _WIN32 @@ -299,14 +300,52 @@ namespace HttpRequestWrapper { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Destroying HTTP request client"); #endif + should_abort_ = true; + int s = active_socket_.exchange(-1); + if (s >= 0) { + #ifdef _WIN32 + ::shutdown(static_cast(s), SD_BOTH); + #else + ::shutdown(s, SHUT_RDWR); + #endif + } + if (request_future_.valid()) { + request_future_.wait(); + } #ifdef WITH_SSL cleanupSSL(); #endif + if (isolate_ != nullptr) { + v8::Locker locker{isolate_}; + v8::Isolate::Scope isolate_scope(isolate_); + v8::HandleScope handle_scope(isolate_); + on_ready_state_change_.Reset(); + on_load_.Reset(); + on_error_.Reset(); + on_timeout_.Reset(); + on_abort_.Reset(); + on_progress_.Reset(); + on_load_start_.Reset(); + on_load_end_.Reset(); + } } void HttpRequestClient::open(const std::string& method, const std::string& url, bool async) { + if (request_future_.valid()) { + should_abort_ = true; + int s = active_socket_.exchange(-1); + if (s >= 0) { + #ifdef _WIN32 + ::shutdown(static_cast(s), SD_BOTH); + #else + ::shutdown(s, SHUT_RDWR); + #endif + } + request_future_.wait(); + } + std::lock_guard lock(state_mutex_); - + method_ = method; url_ = url; async_ = async; @@ -350,10 +389,11 @@ namespace HttpRequestWrapper { #endif request_data_ = data; - + should_abort_ = false; + if (async_) { // use thread pool - GlobalThreadPool::getInstance().enqueue([this]() { + request_future_ = GlobalThreadPool::getInstance().enqueue([this]() { this->performRequest(); }); } else { @@ -364,11 +404,21 @@ namespace HttpRequestWrapper { void HttpRequestClient::abort() { std::lock_guard lock(state_mutex_); - + if (ready_state_ == DONE) { return; } - + + should_abort_ = true; + int s = active_socket_.exchange(-1); + if (s >= 0) { + #ifdef _WIN32 + ::shutdown(static_cast(s), SD_BOTH); + #else + ::shutdown(s, SHUT_RDWR); + #endif + } + setReadyState(DONE); #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "HTTP request aborted"); @@ -437,6 +487,26 @@ namespace HttpRequestWrapper { on_load_end_.Reset(isolate_, callback); } + static size_t response_end(const std::string& response, const std::string& method) { + size_t header_end = response.find("\r\n\r\n"); + if (header_end == std::string::npos) return std::string::npos; + std::string lower(response, 0, header_end); + for (char& c : lower) c = (char)std::tolower((unsigned char)c); + int status = atoi(lower.c_str() + lower.find(' ') + 1); + if (status == 204 || status == 304 || method == "HEAD" || method == "head") { + return header_end + 4; + } + size_t pos = lower.find("content-length:"); + if (pos != std::string::npos) { + return header_end + 4 + strtoull(lower.c_str() + pos + 15, nullptr, 10); + } + if (lower.find("transfer-encoding:") != std::string::npos) { + size_t term = response.find("0\r\n\r\n", header_end + 4); + return term == std::string::npos ? std::string::npos : term + 5; + } + return std::string::npos; + } + void HttpRequestClient::performRequest() { active_operation_ = true; @@ -497,7 +567,8 @@ namespace HttpRequestWrapper { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Acquiring connection for HTTP request"); #endif - SOCKET sock = g_connection_pool.acquire_connection(host, port, is_https); + SOCKET sock = g_connection_pool.acquire_connection(host, port, is_https, &should_abort_); + active_socket_ = static_cast(sock); if (sock == INVALID_SOCKET) { response_.error = "Failed to acquire connection"; #ifdef DEBUG_NETWORK @@ -523,27 +594,36 @@ namespace HttpRequestWrapper { ::setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)); #endif } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Connected successfully to %s:%d", host.c_str(), port); + #endif #ifdef WITH_SSL if (is_https) { + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Performing SSL handshake"); + #endif ssl_socket_ = static_cast(sock); if (!performSSLHandshake(static_cast(sock), host)) { response_.error = "SSL handshake failed"; + active_socket_ = -1; closesocket(sock); setReadyState(DONE); handleError(response_.error); active_operation_ = false; return; } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "SSL handshake completed successfully"); + #endif } #endif std::string request_data = formatHttpRequest(method_, path, host, request_data_); + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Sending HTTP request (%zu bytes):", request_data.length()); kinc_log(KINC_LOG_LEVEL_INFO, "Request preview: %.200s", request_data.c_str()); + #endif int bytes_sent = 0; #ifdef WITH_SSL @@ -565,20 +645,34 @@ namespace HttpRequestWrapper { return; } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "HTTP request sent successfully"); + #endif char buffer[4096]; int bytes_received; std::string response_data; int total_bytes = 0; + bool body_complete = false; + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Starting to receive HTTP response"); + #endif while (true) { + size_t end = response_end(response_data, method_); + if (end != std::string::npos && response_data.size() >= end) { + body_complete = true; + break; + } #ifdef WITH_SSL if (is_https) { + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Calling sslRead for %zu bytes", sizeof(buffer) - 1); + #endif bytes_received = sslRead(buffer, sizeof(buffer) - 1); + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "sslRead returned %d bytes", bytes_received); + #endif } else { bytes_received = ::recv(sock, buffer, sizeof(buffer) - 1, 0); } @@ -587,16 +681,21 @@ namespace HttpRequestWrapper { #endif if (bytes_received <= 0) { + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Breaking from response reading loop, bytes_received: %d", bytes_received); + #endif break; } - buffer[bytes_received] = '\0'; - response_data += buffer; + response_data.append(buffer, bytes_received); total_bytes += bytes_received; + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Received %d bytes (total: %d)", bytes_received, total_bytes); + #endif } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Finished receiving data, total bytes: %d", total_bytes); + #endif #ifdef WITH_SSL if (is_https) { @@ -604,24 +703,40 @@ namespace HttpRequestWrapper { } #endif - // return connection to pool for reuse instead of closing - g_connection_pool.release_connection(host, port, sock, is_https); - + active_socket_ = -1; + if (bytes_received == SOCKET_ERROR) { response_.error = "Failed to receive response"; + closesocket(sock); setReadyState(DONE); handleError(response_.error); active_operation_ = false; return; } + + if (should_abort_.load()) { + closesocket(sock); + active_operation_ = false; + return; + } + + if (body_complete) { + g_connection_pool.release_connection(host, port, sock, is_https); + } else { + closesocket(sock); + } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Raw HTTP response data (%zu bytes):", response_data.length()); kinc_log(KINC_LOG_LEVEL_INFO, "Response preview: %.200s", response_data.c_str()); + #endif parseHttpResponse(response_data); + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Parsed response - status: %d, body length: %zu", response_.status, response_.responseText.length()); kinc_log(KINC_LOG_LEVEL_INFO, "Response body preview: %.200s", response_.responseText.c_str()); + #endif setReadyState(HEADERS_RECEIVED); @@ -661,6 +776,7 @@ namespace HttpRequestWrapper { } void HttpRequestClient::handleError(const std::string& error) { + if (should_abort_.load()) return; // abort() already fired its event std::lock_guard lock(event_queue_mutex_); event_queue_.push(Event(HTTP_ERROR, error)); } @@ -832,7 +948,9 @@ namespace HttpRequestWrapper { bool HttpRequestClient::parseUrl(const std::string& url, std::string& host, int& port, std::string& path, bool& is_https) { if (url.find("://") == std::string::npos) { + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Detected local file path: %s", url.c_str()); + #endif host = "localhost"; port = 0; path = url; @@ -841,7 +959,9 @@ namespace HttpRequestWrapper { } if (url.substr(0, 7) == "file://") { + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Detected file:// URL: %s", url.c_str()); + #endif host = "localhost"; port = 0; // Special port to indicate local file path = url.substr(7); // Remove "file://" prefix @@ -852,7 +972,9 @@ namespace HttpRequestWrapper { std::regex url_regex(R"(^(https?):\/\/([^\/\s:]+)(?::(\d+))?(\/[^\s]*)?(?:\?[^\s]*)?(?:#[^\s]*)?$)"); std::smatch matches; + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Parsing HTTP URL: %s", url.c_str()); + #endif if (!std::regex_match(url, matches, url_regex)) { // fallback parsing for simpler URLs @@ -900,8 +1022,10 @@ namespace HttpRequestWrapper { port = is_https ? 443 : 80; } - kinc_log(KINC_LOG_LEVEL_INFO, "URL parsed successfully - scheme: %s, host: %s, port: %d, path: %s", + #ifdef DEBUG_NETWORK + kinc_log(KINC_LOG_LEVEL_INFO, "URL parsed successfully - scheme: %s, host: %s, port: %d, path: %s", scheme.c_str(), host.c_str(), port, path.c_str()); + #endif return true; } @@ -911,7 +1035,7 @@ namespace HttpRequestWrapper { request << method << " " << path << " HTTP/1.1\r\n"; request << "Host: " << host << "\r\n"; - request << "Connection: close\r\n"; + request << "Connection: keep-alive\r\n"; request << "User-Agent: LNXCORE/1.0\r\n"; request << "Accept: */*\r\n"; request << "Accept-Encoding: identity\r\n"; @@ -935,8 +1059,10 @@ namespace HttpRequestWrapper { } std::string request_str = request.str(); + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Formatted HTTP request (%zu bytes):", request_str.length()); kinc_log(KINC_LOG_LEVEL_INFO, "Request headers: %s", request_str.substr(0, request_str.find("\r\n\r\n")).c_str()); + #endif return request_str; } @@ -1011,11 +1137,15 @@ namespace HttpRequestWrapper { response_.headers[header_name] = header_value; header_count++; + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Header: %s = %s", header_name.c_str(), header_value.c_str()); + #endif } } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Parsed %d headers", header_count); + #endif std::ostringstream body_stream; std::string body_line; @@ -1030,12 +1160,16 @@ namespace HttpRequestWrapper { } response_.responseText = body_stream.str(); + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Parsed body: %zu lines, %zu bytes", body_lines, response_.responseText.length()); + #endif + #ifdef DEBUG_NETWORK if (!response_.responseText.empty()) { std::string preview = response_.responseText.substr(0, 100); kinc_log(KINC_LOG_LEVEL_INFO, "Response body preview: %s", preview.c_str()); } + #endif } void HttpRequestClient::loadLocalFile(const std::string& path) { @@ -2394,4 +2528,4 @@ void bind_httprequest_class(Isolate* isolate, const Global& context) { global->Set(current_context, String::NewFromUtf8(isolate, "XMLHttpRequest").ToLocalChecked(), xhrFunc); } -#endif \ No newline at end of file +#endif \ No newline at end of file diff --git a/Sources/httprequest.h b/Sources/httprequest.h index cd727ab..20e720c 100644 --- a/Sources/httprequest.h +++ b/Sources/httprequest.h @@ -11,6 +11,7 @@ #include #include #include +#include #include #ifdef WITH_SSL @@ -101,7 +102,9 @@ namespace HttpRequestWrapper { bool with_credentials_; bool async_; std::thread request_thread_; - std::atomic should_abort_; + std::atomic should_abort_{false}; + std::future request_future_; + std::atomic active_socket_{-1}; v8::Global on_ready_state_change_; v8::Global on_load_; diff --git a/Sources/lockfree_queue.h b/Sources/lockfree_queue.h index 66303f4..a06274e 100644 --- a/Sources/lockfree_queue.h +++ b/Sources/lockfree_queue.h @@ -59,6 +59,7 @@ public: } result = *data; + next->data.store(nullptr); delete data; head_.store(next); diff --git a/Sources/main.cpp b/Sources/main.cpp index 264b991..42723a0 100644 --- a/Sources/main.cpp +++ b/Sources/main.cpp @@ -38,6 +38,10 @@ #include "websocket.h" #endif +#ifdef WITH_RTC +#include "rtc_v8_bindings.h" +#endif + #ifdef WITH_BENCHMARK #include "BENCHMARK/websocket_benchmark.h" extern "C" void websocket_run_benchmark(); @@ -1325,9 +1329,7 @@ namespace { } int runt_get_samples_per_second_fast(Local receiver) { - int rate = kinc_a2_samples_per_second(); - kinc_log(KINC_LOG_LEVEL_INFO, "Samples per second: %d Hz.", rate); - return rate; + return kinc_a2_samples_per_second(); } void runt_get_samples_per_second(const FunctionCallbackInfo &args) { @@ -2296,11 +2298,14 @@ namespace { HandleScope scope(args.GetIsolate()); String::Utf8Value utf8_path(isolate, args[0]); + if (args.Length() < 2 || !args[1]->IsArrayBuffer()) return; + Local buffer = Local::Cast(args[1]); std::shared_ptr content = buffer->GetBackingStore(); bool hasLengthArg = args.Length() > 2 && !args[2]->IsNullOrUndefined(); int byteLength = hasLengthArg ? args[2]->ToInt32(isolate->GetCurrentContext()).ToLocalChecked()->Value() : (int)content->ByteLength(); + if (byteLength < 0) byteLength = 0; if (byteLength > (int)content->ByteLength()) byteLength = (int)content->ByteLength(); #ifdef KINC_WINDOWS @@ -2314,6 +2319,112 @@ namespace { fclose(file); } + void runt_file_write_bytes(const FunctionCallbackInfo &args) { + HandleScope scope(args.GetIsolate()); + String::Utf8Value utf8_path(isolate, args[0]); + + if (args.Length() < 2 || !args[1]->IsArrayBuffer()) return; + + Local buffer = Local::Cast(args[1]); + std::shared_ptr content = buffer->GetBackingStore(); + + int64_t offset = 0; + if (args.Length() > 2 && !args[2]->IsNullOrUndefined()) { + offset = (int64_t)args[2]->ToNumber(isolate->GetCurrentContext()).ToLocalChecked()->Value(); + } + + bool hasLengthArg = args.Length() > 3 && !args[3]->IsNullOrUndefined(); + int byteLength = hasLengthArg ? args[3]->ToInt32(isolate->GetCurrentContext()).ToLocalChecked()->Value() : (int)content->ByteLength(); + if (byteLength < 0) byteLength = 0; + if (byteLength > (int)content->ByteLength()) byteLength = (int)content->ByteLength(); + + #ifdef KINC_WINDOWS + MultiByteToWideChar(CP_UTF8, 0, *utf8_path, -1, temp_wstring, 1024); + FILE *file = _wfopen(temp_wstring, L"r+b"); + if (file == nullptr) file = _wfopen(temp_wstring, L"w+b"); + #else + FILE *file = fopen(*utf8_path, "r+b"); + if (file == nullptr) file = fopen(*utf8_path, "w+b"); + #endif + if (file == nullptr) { + args.GetReturnValue().Set(Boolean::New(isolate, false)); + return; + } + #ifdef KINC_WINDOWS + _fseeki64(file, offset, SEEK_SET); + #else + fseeko(file, offset, SEEK_SET); + #endif + size_t written = fwrite(content->Data(), 1, byteLength, file); + fclose(file); + args.GetReturnValue().Set(Boolean::New(isolate, written == (size_t)byteLength)); + } + + void runt_file_read_bytes(const FunctionCallbackInfo &args) { + HandleScope scope(args.GetIsolate()); + String::Utf8Value utf8_path(isolate, args[0]); + + int64_t offset = 0; + if (args.Length() > 1 && !args[1]->IsNullOrUndefined()) { + offset = (int64_t)args[1]->ToNumber(isolate->GetCurrentContext()).ToLocalChecked()->Value(); + } + int length = 0; + if (args.Length() > 2 && !args[2]->IsNullOrUndefined()) { + length = args[2]->ToInt32(isolate->GetCurrentContext()).ToLocalChecked()->Value(); + } + if (offset < 0 || length <= 0) { + args.GetReturnValue().Set(ArrayBuffer::New(isolate, 0)); + return; + } + + Local buffer = ArrayBuffer::New(isolate, length); + std::shared_ptr content = buffer->GetBackingStore(); + memset(content->Data(), 0, length); + + #ifdef KINC_WINDOWS + MultiByteToWideChar(CP_UTF8, 0, *utf8_path, -1, temp_wstring, 1024); + FILE *file = _wfopen(temp_wstring, L"rb"); + #else + FILE *file = fopen(*utf8_path, "rb"); + #endif + if (file != nullptr) { + #ifdef KINC_WINDOWS + _fseeki64(file, offset, SEEK_SET); + #else + fseeko(file, offset, SEEK_SET); + #endif + fread(content->Data(), 1, length, file); + fclose(file); + } + + args.GetReturnValue().Set(buffer); + } + + void runt_file_size(const FunctionCallbackInfo &args) { + HandleScope scope(args.GetIsolate()); + String::Utf8Value utf8_path(isolate, args[0]); + int64_t size = -1; + + #ifdef KINC_WINDOWS + MultiByteToWideChar(CP_UTF8, 0, *utf8_path, -1, temp_wstring, 1024); + FILE *file = _wfopen(temp_wstring, L"rb"); + #else + FILE *file = fopen(*utf8_path, "rb"); + #endif + if (file != nullptr) { + #ifdef KINC_WINDOWS + _fseeki64(file, 0, SEEK_END); + size = _ftelli64(file); + #else + fseeko(file, 0, SEEK_END); + size = (int64_t)ftello(file); + #endif + fclose(file); + } + + args.GetReturnValue().Set(Number::New(isolate, (double)size)); + } + int sys_command(const char *cmd) { #ifdef KINC_WINDOWS int wlen = MultiByteToWideChar(CP_UTF8, 0, cmd, -1, NULL, 0); @@ -2913,6 +3024,9 @@ namespace { SET_FUNCTION(runt, "beginFace", runt_begin_face); SET_FUNCTION(runt, "end", runt_end); SET_FUNCTION(runt, "fileSaveBytes", runt_file_save_bytes); + SET_FUNCTION(runt, "fileWriteBytes", runt_file_write_bytes); + SET_FUNCTION(runt, "fileReadBytes", runt_file_read_bytes); + SET_FUNCTION(runt, "fileSize", runt_file_size); SET_FUNCTION(runt, "sysCommand", runt_sys_command); SET_FUNCTION(runt, "savePath", runt_save_path); SET_FUNCTION(runt, "getArgCount", runt_get_arg_count); @@ -2934,6 +3048,10 @@ namespace { SET_FUNCTION(runt, "socketSetBlocking", v8_runt_socket_set_blocking); SET_FUNCTION(runt, "socketSelect", v8_runt_socket_select); SET_FUNCTION(runt, "socketIsConnected", v8_runt_socket_is_connected); + SET_FUNCTION(runt, "udpCreate", v8_runt_udp_create); + SET_FUNCTION(runt, "udpBind", v8_runt_udp_bind); + SET_FUNCTION(runt, "udpSendTo", v8_runt_udp_sendto); + SET_FUNCTION(runt, "udpRecvFrom", v8_runt_udp_recvfrom); #ifdef WITH_SSL SET_FUNCTION(runt, "socketEnableSsl", v8_runt_socket_enable_ssl); #endif @@ -3214,6 +3332,9 @@ namespace { WebSocketWrapper::processEvents(); HttpRequestWrapper::processEvents(); #endif + #ifdef WITH_RTC + runt_rtc_process_events_native(); + #endif execute_timers(); EngineManager::AsyncEngine::instance().process_events(); @@ -3454,10 +3575,14 @@ namespace { if (!func->Call(context, context->Global(), 0, NULL).ToLocal(&result)) { handle_exception(&try_catch); } - + + #ifdef WITH_RTC + v8_runt_rtc_shutdown_all(); + #endif #ifdef WITH_NETWORKING WebSocketWrapper::cleanup(); HttpRequestWrapper::cleanup(); + GlobalThreadPool::shutdown(); #endif #ifdef WITH_VIEWPORT if (viewport_server_mode) { @@ -4115,6 +4240,9 @@ int kickstart(int argc, char **argv) { bind_httprequest_class(isolate, global_context); WEBSOCKET_BIND_V8(isolate, global_context); #endif + #ifdef WITH_RTC + bind_rtc_bridge(isolate, global_context); + #endif start_runt(snapshot_found ? NULL : code); diff --git a/Sources/rtc_bridge.cpp b/Sources/rtc_bridge.cpp new file mode 100644 index 0000000..0ef05c7 --- /dev/null +++ b/Sources/rtc_bridge.cpp @@ -0,0 +1,232 @@ +#include "rtc_bridge.h" + +#include + +#include + +#ifdef _WIN32 +#include +#include +#else +#include +#endif + +static std::mutex *g_rtc_mutex = new std::mutex; +static std::deque *g_rtc_events = new std::deque; + +static void rtc_push(RtcEvent &&ev) { + std::lock_guard lock(*g_rtc_mutex); + g_rtc_events->push_back(std::move(ev)); +} + +bool runt_rtc_poll_event(RtcEvent &out) { + std::lock_guard lock(*g_rtc_mutex); + if (g_rtc_events->empty()) return false; + out = std::move(g_rtc_events->front()); + g_rtc_events->pop_front(); + return true; +} + +static void RTC_API rtc_description_cb(int pc, const char *sdp, const char *type, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::LocalDescription; + ev.id = pc; + ev.s1 = sdp != nullptr ? sdp : ""; + ev.s2 = type != nullptr ? type : ""; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_candidate_cb(int pc, const char *cand, const char *mid, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::LocalCandidate; + ev.id = pc; + ev.s1 = cand != nullptr ? cand : ""; + ev.s2 = mid != nullptr ? mid : ""; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_open_cb(int id, void *ptr); +static void RTC_API rtc_closed_cb(int id, void *ptr); +static void RTC_API rtc_error_cb(int id, const char *error, void *ptr); +static void RTC_API rtc_message_cb(int id, const char *message, int size, void *ptr); + +static void RTC_API rtc_state_cb(int pc, rtcState state, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::StateChange; + ev.id = pc; + ev.arg = (int)state; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_gathering_cb(int pc, rtcGatheringState state, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::GatheringState; + ev.id = pc; + ev.arg = (int)state; + rtc_push(std::move(ev)); +} + +static void rtc_register_dc_callbacks(int dc) { + rtcSetOpenCallback(dc, rtc_open_cb); + rtcSetClosedCallback(dc, rtc_closed_cb); + rtcSetErrorCallback(dc, rtc_error_cb); + rtcSetMessageCallback(dc, rtc_message_cb); +} + +static void RTC_API rtc_datachannel_cb(int pc, int dc, void *ptr) { + rtc_register_dc_callbacks(dc); + RtcEvent ev; + ev.type = RtcEventType::DataChannel; + ev.id = pc; + ev.arg = dc; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_open_cb(int id, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::Open; + ev.id = id; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_closed_cb(int id, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::Closed; + ev.id = id; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_error_cb(int id, const char *error, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::Error; + ev.id = id; + ev.s1 = error != nullptr ? error : ""; + rtc_push(std::move(ev)); +} + +static void RTC_API rtc_message_cb(int id, const char *message, int size, void *ptr) { + RtcEvent ev; + ev.type = RtcEventType::Message; + ev.id = id; + if (size < 0) { + ev.isString = true; + ev.s1 = message != nullptr ? message : ""; + } else { + ev.data.assign(message, message + size); + } + rtc_push(std::move(ev)); +} + +int runt_rtc_init() { + return rtcPreload() ? 0 : -1; +} + +void runt_rtc_cleanup() { + rtcCleanup(); +} + +int runt_rtc_create_peer_connection(const std::vector &iceServers) { + rtcConfiguration conf; + memset(&conf, 0, sizeof(conf)); + std::vector servers(iceServers.size()); + for (size_t i = 0; i < iceServers.size(); i++) { + servers[i] = iceServers[i].c_str(); + } + conf.iceServers = servers.data(); + conf.iceServersCount = (int)servers.size(); + int pc = rtcCreatePeerConnection(&conf); + if (pc < 0) return pc; + rtcSetLocalDescriptionCallback(pc, rtc_description_cb); + rtcSetLocalCandidateCallback(pc, rtc_candidate_cb); + rtcSetStateChangeCallback(pc, rtc_state_cb); + rtcSetGatheringStateChangeCallback(pc, rtc_gathering_cb); + rtcSetDataChannelCallback(pc, rtc_datachannel_cb); + return pc; +} + +void runt_rtc_close_peer_connection(int pc) { + rtcClosePeerConnection(pc); +} + +void runt_rtc_delete_peer_connection(int pc) { + rtcDeletePeerConnection(pc); +} + +void runt_rtc_set_local_description(int pc, const char *type) { + rtcSetLocalDescription(pc, type); +} + +int runt_rtc_set_remote_description(int pc, const char *sdp, const char *type) { + return rtcSetRemoteDescription(pc, sdp, type); +} + +int runt_rtc_add_remote_candidate(int pc, const char *cand, const char *mid) { + return rtcAddRemoteCandidate(pc, cand, mid); +} + +std::string runt_rtc_get_local_description(int pc) { + int size = rtcGetLocalDescription(pc, nullptr, 0); + if (size <= 0) return ""; + std::string buf(size, '\0'); + rtcGetLocalDescription(pc, buf.data(), size); + return std::string(buf.data(), strnlen(buf.data(), buf.size())); +} + +std::string runt_rtc_get_local_description_type(int pc) { + char buf[64]; + int size = rtcGetLocalDescriptionType(pc, buf, sizeof(buf)); + if (size <= 0) return ""; + return std::string(buf, strnlen(buf, sizeof(buf))); +} + +int runt_rtc_create_data_channel(int pc, const char *label, bool unordered, + int maxRetransmits) { + rtcDataChannelInit init; + memset(&init, 0, sizeof(init)); + init.reliability.unordered = unordered; + init.reliability.unreliable = maxRetransmits >= 0; + init.reliability.maxRetransmits = + maxRetransmits >= 0 ? (unsigned int)maxRetransmits : 0; + int dc = rtcCreateDataChannelEx(pc, label, &init); + if (dc < 0) return dc; + rtc_register_dc_callbacks(dc); + return dc; +} + +void runt_rtc_delete_data_channel(int dc) { + rtcDeleteDataChannel(dc); +} + +int runt_rtc_send_message(int id, const char *data, int size) { + return rtcSendMessage(id, data, size); +} + +void runt_rtc_close(int id) { + rtcClose(id); +} + +void runt_rtc_delete(int id) { + rtcDelete(id); +} + +bool runt_rtc_is_open(int id) { + return rtcIsOpen(id); +} + +int runt_rtc_get_buffered_amount(int id) { + return rtcGetBufferedAmount(id); +} + +bool runt_rtc_random_bytes(char *out, int len) { +#ifdef _WIN32 + return BCryptGenRandom(nullptr, reinterpret_cast(out), + static_cast(len), + BCRYPT_USE_SYSTEM_PREFERRED_RNG) == 0; +#else + FILE *f = fopen("/dev/urandom", "rb"); + if (f == nullptr) return false; + size_t n = fread(out, 1, static_cast(len), f); + fclose(f); + return n == static_cast(len); +#endif +} diff --git a/Sources/rtc_bridge.h b/Sources/rtc_bridge.h new file mode 100644 index 0000000..c98eb66 --- /dev/null +++ b/Sources/rtc_bridge.h @@ -0,0 +1,54 @@ +#pragma once + +#include +#include +#include +#include + +enum class RtcEventType { + LocalDescription, // id=pc, s1=sdp, s2=type + LocalCandidate, // id=pc, s1=cand, s2=mid + StateChange, // id=pc, arg=rtcState + GatheringState, // id=pc, arg=rtcGatheringState + DataChannel, // id=pc, arg=new dc id + Open, // id=dc + Closed, // id=dc + Error, // id=dc, s1=error + Message // id=dc, data=payload, isString=text frame +}; + +struct RtcEvent { + RtcEventType type; + int id; + int arg = 0; + std::string s1; + std::string s2; + std::vector data; + bool isString = false; +}; + +int runt_rtc_init(); +void runt_rtc_cleanup(); + +int runt_rtc_create_peer_connection(const std::vector &iceServers); +void runt_rtc_close_peer_connection(int pc); +void runt_rtc_delete_peer_connection(int pc); +void runt_rtc_set_local_description(int pc, const char *type); +int runt_rtc_set_remote_description(int pc, const char *sdp, const char *type); +int runt_rtc_add_remote_candidate(int pc, const char *cand, const char *mid); +std::string runt_rtc_get_local_description(int pc); +std::string runt_rtc_get_local_description_type(int pc); + +int runt_rtc_create_data_channel(int pc, const char *label, bool unordered, + int maxRetransmits); +void runt_rtc_delete_data_channel(int dc); + +int runt_rtc_send_message(int id, const char *data, int size); +void runt_rtc_close(int id); +void runt_rtc_delete(int id); +bool runt_rtc_is_open(int id); +int runt_rtc_get_buffered_amount(int id); + +bool runt_rtc_random_bytes(char *out, int len); + +bool runt_rtc_poll_event(RtcEvent &out); diff --git a/Sources/rtc_v8_bindings.cpp b/Sources/rtc_v8_bindings.cpp new file mode 100644 index 0000000..e970e61 --- /dev/null +++ b/Sources/rtc_v8_bindings.cpp @@ -0,0 +1,607 @@ +#include "rtc_v8_bindings.h" +#include "rtc_bridge.h" + +#include + +#include +#include +#include +#include +#include + +using namespace v8; + +static Isolate *g_rtc_isolate = nullptr; +static std::map> &g_desc_cbs = *new std::map>; +static std::map> &g_cand_cbs = *new std::map>; +static std::map> &g_state_cbs = *new std::map>; +static std::map> &g_gathering_cbs = *new std::map>; +static std::map> &g_dc_cbs = *new std::map>; +static std::map> &g_open_cbs = *new std::map>; +static std::map> &g_closed_cbs = *new std::map>; +static std::map> &g_error_cbs = *new std::map>; +static std::map> &g_msg_cbs = *new std::map>; + +static std::map> &g_pc_dcs = *new std::map>; +static std::map &g_dc_to_pc = *new std::map; +static Global *g_rtc_context = nullptr; + +static void rtc_throw(Isolate *isolate, const char *msg) { + isolate->ThrowException( + Exception::TypeError(String::NewFromUtf8(isolate, msg).ToLocalChecked())); +} + +static bool rtc_store_cb(const FunctionCallbackInfo &args, + std::map> &cbs, + const char *name) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 2 || !args[0]->IsNumber() || !args[1]->IsFunction()) { + rtc_throw(isolate, name); + return false; + } + int id = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + cbs[id].Reset(isolate, args[1].As()); + return true; +} + +static void rtc_drop_cbs(int id) { + g_desc_cbs.erase(id); + g_cand_cbs.erase(id); + g_state_cbs.erase(id); + g_gathering_cbs.erase(id); + g_dc_cbs.erase(id); + g_open_cbs.erase(id); + g_closed_cbs.erase(id); + g_error_cbs.erase(id); + g_msg_cbs.erase(id); +} + +static void rtc_track_dc(int pc, int dc) { + g_pc_dcs[pc].push_back(dc); + g_dc_to_pc[dc] = pc; +} + +static void rtc_forget_dc(int dc) { + auto it = g_dc_to_pc.find(dc); + if (it == g_dc_to_pc.end()) return; + auto &dcs = g_pc_dcs[it->second]; + dcs.erase(std::remove(dcs.begin(), dcs.end(), dc), dcs.end()); + g_dc_to_pc.erase(it); +} + +static void rtc_call(Persistent &cb, int argc, Local *argv) { + if (cb.IsEmpty()) return; + Isolate *isolate = g_rtc_isolate; + Local ctx = isolate->GetCurrentContext(); + Local fn = cb.Get(isolate); + TryCatch try_catch(isolate); + fn->Call(ctx, Null(isolate), argc, argv).IsEmpty(); + if (try_catch.HasCaught()) { + String::Utf8Value err(isolate, try_catch.Exception()); + kinc_log(KINC_LOG_LEVEL_WARNING, "runt_rtc callback threw: %s", + *err != nullptr ? *err : "unknown"); + } +} + +static Local rtc_str(Isolate *isolate, const std::string &s) { + return String::NewFromUtf8(isolate, s.data(), NewStringType::kNormal, + (int)s.size()) + .ToLocalChecked(); +} + +void v8_runt_rtc_init(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + g_rtc_isolate = isolate; + args.GetReturnValue().Set(Integer::New(isolate, runt_rtc_init())); +} + +void v8_runt_rtc_cleanup(const FunctionCallbackInfo &args) { + runt_rtc_cleanup(); + for (auto *m : {&g_desc_cbs, &g_cand_cbs, &g_state_cbs, + &g_gathering_cbs, &g_dc_cbs, &g_open_cbs, + &g_closed_cbs, &g_error_cbs, &g_msg_cbs}) { + for (auto &kv : *m) kv.second.Reset(); + m->clear(); + } + g_pc_dcs.clear(); + g_dc_to_pc.clear(); + RtcEvent ev; + while (runt_rtc_poll_event(ev)) {} + args.GetReturnValue().Set(Undefined(args.GetIsolate())); +} + +void v8_runt_rtc_shutdown_all() { + std::set dcs; + for (auto &kv : g_dc_to_pc) dcs.insert(kv.first); + for (auto &kv : g_msg_cbs) dcs.insert(kv.first); + for (auto &kv : g_open_cbs) dcs.insert(kv.first); + for (auto &kv : g_closed_cbs) dcs.insert(kv.first); + for (auto &kv : g_error_cbs) dcs.insert(kv.first); + for (int dc : dcs) runt_rtc_delete_data_channel(dc); + + std::set pcs; + for (auto &kv : g_pc_dcs) pcs.insert(kv.first); + for (auto &kv : g_desc_cbs) pcs.insert(kv.first); + for (auto &kv : g_cand_cbs) pcs.insert(kv.first); + for (auto &kv : g_state_cbs) pcs.insert(kv.first); + for (auto &kv : g_gathering_cbs) pcs.insert(kv.first); + for (auto &kv : g_dc_cbs) pcs.insert(kv.first); + for (int pc : pcs) runt_rtc_delete_peer_connection(pc); + + for (auto *m : {&g_desc_cbs, &g_cand_cbs, &g_state_cbs, + &g_gathering_cbs, &g_dc_cbs, &g_open_cbs, + &g_closed_cbs, &g_error_cbs, &g_msg_cbs}) { + for (auto &kv : *m) kv.second.Reset(); + m->clear(); + } + g_pc_dcs.clear(); + g_dc_to_pc.clear(); + RtcEvent ev; + while (runt_rtc_poll_event(ev)) {} + runt_rtc_cleanup(); +} + +void v8_runt_rtc_create_peer_connection(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + std::vector iceServers; + if (args.Length() >= 1 && args[0]->IsArray()) { + Local arr = args[0].As(); + Local ctx = isolate->GetCurrentContext(); + for (uint32_t i = 0; i < arr->Length(); i++) { + Local el = arr->Get(ctx, i).ToLocalChecked(); + if (el->IsString()) { + String::Utf8Value s(isolate, el); + iceServers.push_back(*s); + } + } + } + int pc = runt_rtc_create_peer_connection(iceServers); + args.GetReturnValue().Set(Integer::New(isolate, pc)); +} + +void v8_runt_rtc_close_peer_connection(const FunctionCallbackInfo &args) { + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int pc = args[0]->Int32Value(args.GetIsolate()->GetCurrentContext()).FromJust(); + runt_rtc_close_peer_connection(pc); +} + +void v8_runt_rtc_delete_peer_connection(const FunctionCallbackInfo &args) { + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int pc = args[0]->Int32Value(args.GetIsolate()->GetCurrentContext()).FromJust(); + runt_rtc_delete_peer_connection(pc); + rtc_drop_cbs(pc); + auto it = g_pc_dcs.find(pc); + if (it != g_pc_dcs.end()) { + for (int dc : it->second) { + rtc_drop_cbs(dc); + g_dc_to_pc.erase(dc); + } + g_pc_dcs.erase(it); + } +} + +void v8_runt_rtc_set_local_description(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int pc = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + const char *type = nullptr; + if (args.Length() >= 2 && args[1]->IsString()) { + String::Utf8Value typeStr(isolate, args[1]); + type = *typeStr; + } + runt_rtc_set_local_description(pc, type); +} + +void v8_runt_rtc_set_remote_description(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 3 || !args[0]->IsNumber() || !args[1]->IsString() || + !args[2]->IsString()) { + rtc_throw(isolate, + "runt_rtc_set_remote_description requires (pc, sdp, type)"); + return; + } + int pc = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + String::Utf8Value sdp(isolate, args[1]); + String::Utf8Value type(isolate, args[2]); + int ret = runt_rtc_set_remote_description(pc, *sdp, *type); + args.GetReturnValue().Set(Integer::New(isolate, ret)); +} + +void v8_runt_rtc_add_remote_candidate(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 3 || !args[0]->IsNumber() || !args[1]->IsString() || + !args[2]->IsString()) { + rtc_throw(isolate, + "runt_rtc_add_remote_candidate requires (pc, cand, mid)"); + return; + } + int pc = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + String::Utf8Value cand(isolate, args[1]); + String::Utf8Value mid(isolate, args[2]); + int ret = runt_rtc_add_remote_candidate(pc, *cand, *mid); + args.GetReturnValue().Set(Integer::New(isolate, ret)); +} + +void v8_runt_rtc_get_local_description(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int pc = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + args.GetReturnValue().Set( + rtc_str(isolate, runt_rtc_get_local_description(pc))); +} + +void v8_runt_rtc_get_local_description_type( + const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int pc = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + args.GetReturnValue().Set( + rtc_str(isolate, runt_rtc_get_local_description_type(pc))); +} + +void v8_runt_rtc_create_data_channel(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 2 || !args[0]->IsNumber() || !args[1]->IsString()) { + rtc_throw(isolate, "runt_rtc_create_data_channel requires (pc, label)"); + return; + } + int pc = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + String::Utf8Value label(isolate, args[1]); + bool unordered = false; + int maxRetransmits = -1; + if (args.Length() >= 3 && args[2]->IsBoolean()) { + unordered = args[2]->BooleanValue(isolate); + } + if (args.Length() >= 4 && args[3]->IsNumber()) { + maxRetransmits = + args[3]->Int32Value(isolate->GetCurrentContext()).FromJust(); + } + int dc = runt_rtc_create_data_channel(pc, *label, unordered, + maxRetransmits); + if (dc >= 0) rtc_track_dc(pc, dc); + args.GetReturnValue().Set(Integer::New(isolate, dc)); +} + +void v8_runt_rtc_delete_data_channel(const FunctionCallbackInfo &args) { + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int dc = args[0]->Int32Value(args.GetIsolate()->GetCurrentContext()).FromJust(); + runt_rtc_delete_data_channel(dc); + rtc_drop_cbs(dc); + rtc_forget_dc(dc); +} + +void v8_runt_rtc_send_message(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 2 || !args[0]->IsNumber()) { + rtc_throw(isolate, "runt_rtc_send_message requires (id, data)"); + return; + } + int id = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + int ret = -1; + if (args[1]->IsArrayBuffer()) { + Local buf = args[1].As(); + ret = runt_rtc_send_message( + id, static_cast(buf->GetBackingStore()->Data()), + (int)buf->ByteLength()); + } else if (args[1]->IsArrayBufferView()) { + Local view = args[1].As(); + Local buf = view->Buffer(); + const char *data = + static_cast(buf->GetBackingStore()->Data()) + + view->ByteOffset(); + ret = runt_rtc_send_message(id, data, (int)view->ByteLength()); + } else if (args[1]->IsString()) { + String::Utf8Value s(isolate, args[1]); + ret = runt_rtc_send_message(id, *s, -1); + } else { + rtc_throw(isolate, + "runt_rtc_send_message data must be String or ArrayBuffer"); + return; + } + args.GetReturnValue().Set(Integer::New(isolate, ret)); +} + +void v8_runt_rtc_close(const FunctionCallbackInfo &args) { + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int id = args[0]->Int32Value(args.GetIsolate()->GetCurrentContext()).FromJust(); + runt_rtc_close(id); +} + +void v8_runt_rtc_delete(const FunctionCallbackInfo &args) { + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int id = args[0]->Int32Value(args.GetIsolate()->GetCurrentContext()).FromJust(); + runt_rtc_delete(id); + rtc_drop_cbs(id); + rtc_forget_dc(id); +} + +void v8_runt_rtc_is_open(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int id = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + args.GetReturnValue().Set(Boolean::New(isolate, runt_rtc_is_open(id))); +} + +void v8_runt_rtc_get_buffered_amount(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 1 || !args[0]->IsNumber()) return; + int id = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + args.GetReturnValue().Set( + Integer::New(isolate, runt_rtc_get_buffered_amount(id))); +} + +void v8_runt_rtc_set_local_description_callback( + const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_desc_cbs, + "runt_rtc_set_local_description_callback requires (pc, fn)"); +} + +void v8_runt_rtc_set_local_candidate_callback( + const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_cand_cbs, + "runt_rtc_set_local_candidate_callback requires (pc, fn)"); +} + +void v8_runt_rtc_set_state_change_callback( + const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_state_cbs, + "runt_rtc_set_state_change_callback requires (pc, fn)"); +} + +void v8_runt_rtc_set_gathering_state_callback( + const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_gathering_cbs, + "runt_rtc_set_gathering_state_callback requires (pc, fn)"); +} + +void v8_runt_rtc_set_data_channel_callback( + const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_dc_cbs, + "runt_rtc_set_data_channel_callback requires (pc, fn)"); +} + +void v8_runt_rtc_set_open_callback(const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_open_cbs, + "runt_rtc_set_open_callback requires (id, fn)"); +} + +void v8_runt_rtc_set_closed_callback(const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_closed_cbs, + "runt_rtc_set_closed_callback requires (id, fn)"); +} + +void v8_runt_rtc_set_error_callback(const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_error_cbs, + "runt_rtc_set_error_callback requires (id, fn)"); +} + +void v8_runt_rtc_set_message_callback(const FunctionCallbackInfo &args) { + rtc_store_cb(args, g_msg_cbs, + "runt_rtc_set_message_callback requires (id, fn)"); +} + +void v8_runt_rtc_random_bytes(const FunctionCallbackInfo &args) { + Isolate *isolate = args.GetIsolate(); + if (args.Length() < 1 || !args[0]->IsNumber()) { + rtc_throw(isolate, "runt_rtc_random_bytes requires (len)"); + return; + } + int len = args[0]->Int32Value(isolate->GetCurrentContext()).FromJust(); + if (len <= 0) { + args.GetReturnValue().Set(Null(isolate)); + return; + } + Local buf = ArrayBuffer::New(isolate, len); + if (!runt_rtc_random_bytes( + static_cast(buf->GetBackingStore()->Data()), len)) { + args.GetReturnValue().Set(Null(isolate)); + return; + } + args.GetReturnValue().Set(buf); +} + +static void rtc_drain_events() { + Isolate *isolate = g_rtc_isolate; + RtcEvent ev; + while (runt_rtc_poll_event(ev)) { + switch (ev.type) { + case RtcEventType::LocalDescription: { + auto it = g_desc_cbs.find(ev.id); + if (it != g_desc_cbs.end()) { + Local argv[2] = {rtc_str(isolate, ev.s1), + rtc_str(isolate, ev.s2)}; + rtc_call(it->second, 2, argv); + } + break; + } + case RtcEventType::LocalCandidate: { + auto it = g_cand_cbs.find(ev.id); + if (it != g_cand_cbs.end()) { + Local argv[2] = {rtc_str(isolate, ev.s1), + rtc_str(isolate, ev.s2)}; + rtc_call(it->second, 2, argv); + } + break; + } + case RtcEventType::StateChange: { + auto it = g_state_cbs.find(ev.id); + if (it != g_state_cbs.end()) { + Local argv[1] = {Integer::New(isolate, ev.arg)}; + rtc_call(it->second, 1, argv); + } + break; + } + case RtcEventType::GatheringState: { + auto it = g_gathering_cbs.find(ev.id); + if (it != g_gathering_cbs.end()) { + Local argv[1] = {Integer::New(isolate, ev.arg)}; + rtc_call(it->second, 1, argv); + } + break; + } + case RtcEventType::DataChannel: { + rtc_track_dc(ev.id, ev.arg); + auto it = g_dc_cbs.find(ev.id); + if (it != g_dc_cbs.end()) { + Local argv[1] = {Integer::New(isolate, ev.arg)}; + rtc_call(it->second, 1, argv); + } + break; + } + case RtcEventType::Open: { + auto it = g_open_cbs.find(ev.id); + if (it != g_open_cbs.end()) { + rtc_call(it->second, 0, nullptr); + } + break; + } + case RtcEventType::Closed: { + auto it = g_closed_cbs.find(ev.id); + if (it != g_closed_cbs.end()) { + rtc_call(it->second, 0, nullptr); + } + rtc_drop_cbs(ev.id); + rtc_forget_dc(ev.id); + break; + } + case RtcEventType::Error: { + auto it = g_error_cbs.find(ev.id); + if (it != g_error_cbs.end()) { + Local argv[1] = {rtc_str(isolate, ev.s1)}; + rtc_call(it->second, 1, argv); + } + break; + } + case RtcEventType::Message: { + auto it = g_msg_cbs.find(ev.id); + if (it != g_msg_cbs.end()) { + Local arg; + if (ev.isString) { + arg = rtc_str(isolate, ev.s1); + } else { + // move the payload into the backing store (no copy) + auto *payload = new std::vector(std::move(ev.data)); + auto backing = ArrayBuffer::NewBackingStore( + payload->data(), payload->size(), + [](void *, size_t, void *deleter_data) { + delete static_cast *>(deleter_data); + }, + payload); + arg = ArrayBuffer::New(isolate, std::move(backing)); + } + Local argv[1] = {arg}; + rtc_call(it->second, 1, argv); + } + break; + } + } + } +} + +void runt_rtc_process_events_native() { + if (g_rtc_isolate == nullptr || g_rtc_context == nullptr) return; + Locker locker{g_rtc_isolate}; + Isolate::Scope isolate_scope(g_rtc_isolate); + HandleScope handle_scope(g_rtc_isolate); + Local ctx = Local::New(g_rtc_isolate, *g_rtc_context); + Context::Scope context_scope(ctx); + rtc_drain_events(); +} + +void bind_rtc_bridge(Isolate *isolate, const Global &context) { + Locker locker{isolate}; + Isolate::Scope isolate_scope(isolate); + HandleScope handle_scope(isolate); + + Local current_context = Local::New(isolate, context); + Context::Scope context_scope(current_context); + Local global = current_context->Global(); + + g_rtc_isolate = isolate; + if (g_rtc_context == nullptr) { + g_rtc_context = new Global(isolate, context); + } + + Local runt_obj; + Local existing_runt = + global->Get(current_context, + String::NewFromUtf8(isolate, "RunT").ToLocalChecked()) + .ToLocalChecked(); + if (existing_runt->IsObject()) { + runt_obj = existing_runt->ToObject(current_context).ToLocalChecked(); + } else { + runt_obj = Object::New(isolate); + global->Set(current_context, + String::NewFromUtf8(isolate, "RunT").ToLocalChecked(), + runt_obj); + } + + struct RtcBinding { + const char *global_name; + const char *runt_name; + FunctionCallback fn; + }; + static const RtcBinding bindings[] = { + {"runt_rtc_init", "rtcInit", v8_runt_rtc_init}, + {"runt_rtc_cleanup", "rtcCleanup", v8_runt_rtc_cleanup}, + {"runt_rtc_create_peer_connection", "rtcCreatePeerConnection", + v8_runt_rtc_create_peer_connection}, + {"runt_rtc_close_peer_connection", "rtcClosePeerConnection", + v8_runt_rtc_close_peer_connection}, + {"runt_rtc_delete_peer_connection", "rtcDeletePeerConnection", + v8_runt_rtc_delete_peer_connection}, + {"runt_rtc_set_local_description", "rtcSetLocalDescription", + v8_runt_rtc_set_local_description}, + {"runt_rtc_set_remote_description", "rtcSetRemoteDescription", + v8_runt_rtc_set_remote_description}, + {"runt_rtc_add_remote_candidate", "rtcAddRemoteCandidate", + v8_runt_rtc_add_remote_candidate}, + {"runt_rtc_get_local_description", "rtcGetLocalDescription", + v8_runt_rtc_get_local_description}, + {"runt_rtc_get_local_description_type", "rtcGetLocalDescriptionType", + v8_runt_rtc_get_local_description_type}, + {"runt_rtc_create_data_channel", "rtcCreateDataChannel", + v8_runt_rtc_create_data_channel}, + {"runt_rtc_delete_data_channel", "rtcDeleteDataChannel", + v8_runt_rtc_delete_data_channel}, + {"runt_rtc_send_message", "rtcSendMessage", v8_runt_rtc_send_message}, + {"runt_rtc_close", "rtcClose", v8_runt_rtc_close}, + {"runt_rtc_delete", "rtcDelete", v8_runt_rtc_delete}, + {"runt_rtc_is_open", "rtcIsOpen", v8_runt_rtc_is_open}, + {"runt_rtc_get_buffered_amount", "rtcGetBufferedAmount", + v8_runt_rtc_get_buffered_amount}, + {"runt_rtc_set_local_description_callback", + "rtcSetLocalDescriptionCallback", + v8_runt_rtc_set_local_description_callback}, + {"runt_rtc_set_local_candidate_callback", + "rtcSetLocalCandidateCallback", + v8_runt_rtc_set_local_candidate_callback}, + {"runt_rtc_set_state_change_callback", "rtcSetStateChangeCallback", + v8_runt_rtc_set_state_change_callback}, + {"runt_rtc_set_gathering_state_callback", "rtcSetGatheringStateCallback", + v8_runt_rtc_set_gathering_state_callback}, + {"runt_rtc_set_data_channel_callback", "rtcSetDataChannelCallback", + v8_runt_rtc_set_data_channel_callback}, + {"runt_rtc_set_open_callback", "rtcSetOpenCallback", + v8_runt_rtc_set_open_callback}, + {"runt_rtc_set_closed_callback", "rtcSetClosedCallback", + v8_runt_rtc_set_closed_callback}, + {"runt_rtc_set_error_callback", "rtcSetErrorCallback", + v8_runt_rtc_set_error_callback}, + {"runt_rtc_set_message_callback", "rtcSetMessageCallback", + v8_runt_rtc_set_message_callback}, + {"runt_rtc_random_bytes", "rtcRandomBytes", + v8_runt_rtc_random_bytes}, + }; + + for (const RtcBinding &b : bindings) { + Local fn = + Function::New(current_context, b.fn).ToLocalChecked(); + global->Set(current_context, + String::NewFromUtf8(isolate, b.global_name).ToLocalChecked(), + fn); + runt_obj->Set(current_context, + String::NewFromUtf8(isolate, b.runt_name).ToLocalChecked(), + fn); + } +} diff --git a/Sources/rtc_v8_bindings.h b/Sources/rtc_v8_bindings.h new file mode 100644 index 0000000..20c0b23 --- /dev/null +++ b/Sources/rtc_v8_bindings.h @@ -0,0 +1,9 @@ +#pragma once + +#include + +void bind_rtc_bridge(v8::Isolate *isolate, const v8::Global &context); + +void v8_runt_rtc_shutdown_all(); + +void runt_rtc_process_events_native(); diff --git a/Sources/socket_bridge.cpp b/Sources/socket_bridge.cpp index bb1a9ad..f3cad3c 100644 --- a/Sources/socket_bridge.cpp +++ b/Sources/socket_bridge.cpp @@ -44,6 +44,7 @@ public: bool isConnected; bool isBound; bool isListening; + bool connectInProgress; std::string lastError; #ifdef _WIN32 @@ -58,7 +59,8 @@ public: void* ssl_cred_handle_; void* ssl_context_handle_; bool ssl_context_initialized_; - std::vector ssl_buffer_; + std::vector ssl_buffer_; + std::vector ssl_send_buffer_; #else SSL* ssl; SSL_CTX* ssl_ctx; @@ -67,7 +69,7 @@ public: #endif RunTSocket(int socket_id) : id(socket_id), isBlocking(true), isConnected(false), - isBound(false), isListening(false) { + isBound(false), isListening(false), connectInProgress(false) { #ifdef _WIN32 socket_fd = INVALID_SOCKET; #else @@ -245,6 +247,14 @@ extern "C" int runt_socket_create() { int no_sigpipe = 1; setsockopt(socket->socket_fd, SOL_SOCKET, SO_NOSIGPIPE, (const char*)&no_sigpipe, sizeof(no_sigpipe)); #endif +#ifdef _WIN32 + u_long mode = 1; + ioctlsocket(socket->socket_fd, FIONBIO, &mode); +#else + int flags = fcntl(socket->socket_fd, F_GETFL, 0); + fcntl(socket->socket_fd, F_SETFL, flags | O_NONBLOCK); +#endif + socket->isBlocking = false; g_sockets[socket_id] = std::move(socket); #ifdef DEBUG_NETWORK @@ -399,6 +409,20 @@ extern "C" bool runt_socket_connect(int socket_id, const char* hostname, int por connected = true; break; } +#ifdef _WIN32 + int err = WSAGetLastError(); + if (err == WSAEWOULDBLOCK || err == WSAEINPROGRESS) { + connected = true; + socket->connectInProgress = true; + break; + } +#else + if (errno == EINPROGRESS || errno == EWOULDBLOCK) { + connected = true; + socket->connectInProgress = true; + break; + } +#endif } freeaddrinfo(result); @@ -413,7 +437,7 @@ extern "C" bool runt_socket_connect(int socket_id, const char* hostname, int por return false; } - socket->isConnected = true; + socket->isConnected = !socket->connectInProgress; #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "Connected successfully to %s:%d", hostname, port); #endif @@ -447,7 +471,8 @@ extern "C" int runt_socket_send(int socket_id, const char* data, int length) { } int total_size = stream_sizes.cbHeader + length + stream_sizes.cbTrailer; - char* encrypt_buffer = new char[total_size]; + socket->ssl_send_buffer_.resize(total_size); + char* encrypt_buffer = socket->ssl_send_buffer_.data(); SecBufferDesc message_desc; SecBuffer message_buffers[4]; @@ -479,15 +504,12 @@ extern "C" int runt_socket_send(int socket_id, const char* data, int length) { if (status != SEC_E_OK) { kinc_log(KINC_LOG_LEVEL_ERROR, "SChannel encrypt failed: 0x%x", status); - delete[] encrypt_buffer; return -1; } int total_encrypted = message_buffers[0].cbBuffer + message_buffers[1].cbBuffer + message_buffers[2].cbBuffer; int sent = send(socket->socket_fd, encrypt_buffer, total_encrypted, 0); - delete[] encrypt_buffer; - if (sent != total_encrypted) { kinc_log(KINC_LOG_LEVEL_ERROR, "Failed to send encrypted SSL data"); return -1; @@ -644,11 +666,13 @@ extern "C" int runt_socket_recv(int socket_id, int max_length, char** out_data) int received = recv(socket->socket_fd, recv_buffer, sizeof(recv_buffer), 0); if (received <= 0) { delete[] buffer; + if (received < 0 && WSAGetLastError() == WSAEWOULDBLOCK) return 0; if (received == 0) { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "SSL connection closed by server"); #endif - return 0; + socket->isConnected = false; + return -2; } else { kinc_log(KINC_LOG_LEVEL_ERROR, "Failed to receive encrypted data for SSL read"); return -1; @@ -669,11 +693,13 @@ extern "C" int runt_socket_recv(int socket_id, int max_length, char** out_data) int received = recv(socket->socket_fd, recv_buffer, sizeof(recv_buffer), 0); if (received <= 0) { delete[] buffer; + if (received < 0 && WSAGetLastError() == WSAEWOULDBLOCK) return 0; if (received == 0) { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "SSL connection closed by server (no buffered data)"); #endif - return 0; + socket->isConnected = false; + return -2; } else { kinc_log(KINC_LOG_LEVEL_ERROR, "Failed to receive encrypted data"); return -1; @@ -691,6 +717,10 @@ extern "C" int runt_socket_recv(int socket_id, int max_length, char** out_data) if (ssl_error == SSL_ERROR_WANT_READ || ssl_error == SSL_ERROR_WANT_WRITE) { return 0; } + if (ssl_error == SSL_ERROR_ZERO_RETURN) { + socket->isConnected = false; + return -2; + } kinc_log(KINC_LOG_LEVEL_ERROR, "SSL_read failed: %d", ssl_error); return -1; } @@ -755,6 +785,143 @@ extern "C" int runt_socket_recv(int socket_id, int max_length, char** out_data) return received; } +extern "C" int runt_udp_create() { + if (!initialize_networking()) { + kinc_log(KINC_LOG_LEVEL_ERROR, "Failed to initialize networking"); + return -1; + } + + int socket_id = g_next_socket_id++; + auto socket = std::make_unique(socket_id); + +#ifdef _WIN32 + socket->socket_fd = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + if (socket->socket_fd == INVALID_SOCKET) { + kinc_log(KINC_LOG_LEVEL_ERROR, "UDP socket creation failed: %d", + WSAGetLastError()); + return -1; + } +#else + socket->socket_fd = ::socket(AF_INET, SOCK_DGRAM, 0); + if (socket->socket_fd < 0) { + kinc_log(KINC_LOG_LEVEL_ERROR, "UDP socket creation failed: %s", + strerror(errno)); + return -1; + } +#endif + + int reuse = 1; + setsockopt(socket->socket_fd, SOL_SOCKET, SO_REUSEADDR, + (const char*)&reuse, sizeof(reuse)); + + int sndbuf = 256 * 1024; + int rcvbuf = 256 * 1024; + setsockopt(socket->socket_fd, SOL_SOCKET, SO_SNDBUF, + (const char*)&sndbuf, sizeof(sndbuf)); + setsockopt(socket->socket_fd, SOL_SOCKET, SO_RCVBUF, + (const char*)&rcvbuf, sizeof(rcvbuf)); + +#ifdef _WIN32 + u_long mode = 1; + ioctlsocket(socket->socket_fd, FIONBIO, &mode); +#else + int flags = fcntl(socket->socket_fd, F_GETFL, 0); + fcntl(socket->socket_fd, F_SETFL, flags | O_NONBLOCK); +#endif + socket->isBlocking = false; + + g_sockets[socket_id] = std::move(socket); + #ifdef DEBUG_NETWORK + kinc_log(KINC_LOG_LEVEL_INFO, "UDP socket created: ID %d", socket_id); + #endif + return socket_id; +} + +extern "C" bool runt_udp_bind(int socket_id, const char* address, + int port) { + return runt_socket_bind(socket_id, address, port); +} + +extern "C" int runt_udp_sendto(int socket_id, const char* host, int port, + const char* data, int length) { + auto it = g_sockets.find(socket_id); + if (it == g_sockets.end()) return -1; + + auto& socket = it->second; + if (!socket->isValid()) return -1; + + sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + + if (inet_pton(AF_INET, host, &addr.sin_addr) <= 0) { + struct addrinfo hints, *result; + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_DGRAM; + if (getaddrinfo(host, nullptr, &hints, &result) != 0) { + kinc_log(KINC_LOG_LEVEL_ERROR, + "UDP resolve failed: %s", host); + return -1; + } + addr.sin_addr = ((sockaddr_in*)result->ai_addr)->sin_addr; + freeaddrinfo(result); + } + +#ifdef _WIN32 + int sent = sendto(socket->socket_fd, data, length, 0, + (sockaddr*)&addr, sizeof(addr)); + if (sent < 0) { + int error = WSAGetLastError(); + if (error == WSAEWOULDBLOCK) return 0; + kinc_log(KINC_LOG_LEVEL_ERROR, "UDP sendto failed: %d", error); + return -1; + } +#else + int sent = sendto(socket->socket_fd, data, length, MSG_NOSIGNAL, + (sockaddr*)&addr, sizeof(addr)); + if (sent < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) return 0; + kinc_log(KINC_LOG_LEVEL_ERROR, "UDP sendto failed: %s", + strerror(errno)); + return -1; + } +#endif + return sent; +} + +extern "C" int runt_udp_recvfrom(int socket_id, int max_length, + char** out_data, char* out_host, int out_host_len, + int* out_port) { + auto it = g_sockets.find(socket_id); + if (it == g_sockets.end()) return -1; + + auto& socket = it->second; + if (!socket->isValid()) return -1; + + char* buffer = new char[max_length]; + sockaddr_in from; + socklen_t fromlen = sizeof(from); + + int received = recvfrom(socket->socket_fd, buffer, max_length, 0, + (sockaddr*)&from, &fromlen); + if (received < 0) { + delete[] buffer; +#ifdef _WIN32 + if (WSAGetLastError() == WSAEWOULDBLOCK) return 0; +#else + if (errno == EAGAIN || errno == EWOULDBLOCK) return 0; +#endif + return -1; + } + + inet_ntop(AF_INET, &from.sin_addr, out_host, out_host_len); + *out_port = ntohs(from.sin_port); + *out_data = buffer; + return received; +} + extern "C" void runt_socket_close(int socket_id) { auto it = g_sockets.find(socket_id); if (it == g_sockets.end()) return; @@ -767,11 +934,45 @@ extern "C" void runt_socket_close(int socket_id) { } extern "C" bool runt_socket_is_connected(int socket_id) { - auto it = g_sockets.find(socket_id); - if (it == g_sockets.end()) return false; - - auto& socket = it->second; - return socket->isValid() && socket->isConnected; + auto it = g_sockets.find(socket_id); + if (it == g_sockets.end()) return false; + + auto& socket = it->second; + if (!socket->isValid()) return false; + if (socket->isConnected) return true; + if (!socket->connectInProgress) return false; + + fd_set wfds, efds; + FD_ZERO(&wfds); + FD_ZERO(&efds); + FD_SET(socket->socket_fd, &wfds); + FD_SET(socket->socket_fd, &efds); + struct timeval tv = {0, 0}; + + int sel; +#ifdef _WIN32 + sel = ::select(0, nullptr, &wfds, &efds, &tv); +#else + sel = ::select(socket->socket_fd + 1, nullptr, &wfds, &efds, &tv); +#endif + if (sel <= 0 || (!FD_ISSET(socket->socket_fd, &wfds) && + !FD_ISSET(socket->socket_fd, &efds))) { + return false; + } + + int err = 0; + socklen_t errlen = sizeof(err); + if (getsockopt(socket->socket_fd, SOL_SOCKET, SO_ERROR, (char*)&err, &errlen) != 0) { + socket->connectInProgress = false; + return false; + } + + socket->connectInProgress = false; + if (err == 0) { + socket->isConnected = true; + return true; + } + return false; } extern "C" void runt_socket_set_blocking(int socket_id, bool blocking) { @@ -925,7 +1126,10 @@ extern "C" bool runt_socket_enable_ssl(int socket_id) { SSL_set_fd(socket->ssl, socket->socket_fd); + int flags = fcntl(socket->socket_fd, F_GETFL, 0); + fcntl(socket->socket_fd, F_SETFL, flags & ~O_NONBLOCK); int ret = SSL_connect(socket->ssl); + fcntl(socket->socket_fd, F_SETFL, flags); if (ret <= 0) { int ssl_error = SSL_get_error(socket->ssl, ret); kinc_log(KINC_LOG_LEVEL_ERROR, "SSL_connect failed: %d", ssl_error); diff --git a/Sources/socket_bridge.h b/Sources/socket_bridge.h index 464a956..89cd17e 100644 --- a/Sources/socket_bridge.h +++ b/Sources/socket_bridge.h @@ -15,6 +15,13 @@ int runt_socket_accept(int socket_id); bool runt_socket_connect(int socket_id, const char* hostname, int port); int runt_socket_send(int socket_id, const char* data, int length); int runt_socket_recv(int socket_id, int max_length, char** out_data); + +int runt_udp_create(); +bool runt_udp_bind(int socket_id, const char* address, int port); +int runt_udp_sendto(int socket_id, const char* host, int port, + const char* data, int length); +int runt_udp_recvfrom(int socket_id, int max_length, char** out_data, + char* out_host, int out_host_len, int* out_port); #ifdef WITH_SSL bool runt_socket_enable_ssl(int socket_id); #endif diff --git a/Sources/socket_v8_bindings.cpp b/Sources/socket_v8_bindings.cpp index 71235af..c9cdc62 100644 --- a/Sources/socket_v8_bindings.cpp +++ b/Sources/socket_v8_bindings.cpp @@ -235,6 +235,141 @@ void v8_runt_socket_is_connected(const FunctionCallbackInfo& args) { args.GetReturnValue().Set(Boolean::New(isolate, connected)); } +void v8_runt_udp_create(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + + int socket_id = runt_udp_create(); + #ifdef DEBUG_NETWORK + kinc_log(KINC_LOG_LEVEL_INFO, "[V8] runt_udp_create returned: %d", + socket_id); + #endif + + args.GetReturnValue().Set(Integer::New(isolate, socket_id)); +} + +void v8_runt_udp_bind(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + + if (args.Length() < 3 || !args[0]->IsInt32() || !args[1]->IsString() + || !args[2]->IsInt32()) { + isolate->ThrowException(Exception::TypeError( + String::NewFromUtf8(isolate, "Wrong arguments") + .ToLocalChecked())); + return; + } + + int socket_id = args[0]->Int32Value(isolate->GetCurrentContext()) + .FromJust(); + String::Utf8Value address(isolate, args[1]); + int port = args[2]->Int32Value(isolate->GetCurrentContext()) + .FromJust(); + + bool result = runt_udp_bind(socket_id, *address, port); + args.GetReturnValue().Set(Boolean::New(isolate, result)); +} + +void v8_runt_udp_sendto(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + + if (args.Length() < 4 || !args[0]->IsInt32() || !args[1]->IsString() + || !args[2]->IsInt32()) { + isolate->ThrowException(Exception::TypeError( + String::NewFromUtf8(isolate, "Wrong arguments") + .ToLocalChecked())); + return; + } + + int socket_id = args[0]->Int32Value(isolate->GetCurrentContext()) + .FromJust(); + String::Utf8Value host(isolate, args[1]); + int port = args[2]->Int32Value(isolate->GetCurrentContext()) + .FromJust(); + int sent = -1; + + if (args[3]->IsArrayBuffer()) { + Local buffer = Local::Cast(args[3]); + std::shared_ptr backing = + buffer->GetBackingStore(); + sent = runt_udp_sendto(socket_id, *host, port, + static_cast(backing->Data()), + static_cast(backing->ByteLength())); + } else if (args[3]->IsArrayBufferView()) { + Local view = + Local::Cast(args[3]); + Local buffer = view->Buffer(); + std::shared_ptr backing = + buffer->GetBackingStore(); + const char* data = + static_cast(backing->Data()) + + view->ByteOffset(); + sent = runt_udp_sendto(socket_id, *host, port, data, + static_cast(view->ByteLength())); + } else if (args[3]->IsString()) { + String::Utf8Value str_data(isolate, args[3]); + if (*str_data == nullptr) { + args.GetReturnValue().Set(Integer::New(isolate, -1)); + return; + } + sent = runt_udp_sendto(socket_id, *host, port, *str_data, + str_data.length()); + } else { + isolate->ThrowException(Exception::TypeError( + String::NewFromUtf8(isolate, + "Data must be string, ArrayBuffer, or TypedArray") + .ToLocalChecked())); + return; + } + + args.GetReturnValue().Set(Integer::New(isolate, sent)); +} + +void v8_runt_udp_recvfrom(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + Local context = isolate->GetCurrentContext(); + + if (args.Length() < 2 || !args[0]->IsInt32() || !args[1]->IsInt32()) { + isolate->ThrowException(Exception::TypeError( + String::NewFromUtf8(isolate, "Wrong arguments") + .ToLocalChecked())); + return; + } + + int socket_id = args[0]->Int32Value(context).FromJust(); + int max_length = args[1]->Int32Value(context).FromJust(); + + char* data = nullptr; + char host[64] = {0}; + int port = 0; + int received = runt_udp_recvfrom(socket_id, max_length, &data, + host, sizeof(host), &port); + + if (received > 0 && data) { + std::unique_ptr backing = + ArrayBuffer::NewBackingStore(data, received, + [](void* data, size_t length, void* deleter_data) { + delete[] static_cast(data); + }, nullptr); + Local buffer = + ArrayBuffer::New(isolate, std::move(backing)); + + Local obj = Object::New(isolate); + obj->Set(context, + String::NewFromUtf8(isolate, "data").ToLocalChecked(), + buffer); + obj->Set(context, + String::NewFromUtf8(isolate, "host").ToLocalChecked(), + String::NewFromUtf8(isolate, host).ToLocalChecked()); + obj->Set(context, + String::NewFromUtf8(isolate, "port").ToLocalChecked(), + Integer::New(isolate, port)); + args.GetReturnValue().Set(obj); + } else if (received < 0) { + args.GetReturnValue().Set(Integer::New(isolate, -1)); + } else { + args.GetReturnValue().Set(Null(isolate)); + } +} + #ifdef WITH_SSL void v8_runt_socket_enable_ssl(const FunctionCallbackInfo& args) { Isolate* isolate = args.GetIsolate(); @@ -302,6 +437,18 @@ void bind_socket_bridge(Isolate* isolate, const v8::Global& context global->Set(current_context, String::NewFromUtf8(isolate, "runt_socket_is_connected").ToLocalChecked(), Function::New(current_context, v8_runt_socket_is_connected).ToLocalChecked()); + global->Set(current_context, String::NewFromUtf8(isolate, "runt_udp_create").ToLocalChecked(), + Function::New(current_context, v8_runt_udp_create).ToLocalChecked()); + + global->Set(current_context, String::NewFromUtf8(isolate, "runt_udp_bind").ToLocalChecked(), + Function::New(current_context, v8_runt_udp_bind).ToLocalChecked()); + + global->Set(current_context, String::NewFromUtf8(isolate, "runt_udp_sendto").ToLocalChecked(), + Function::New(current_context, v8_runt_udp_sendto).ToLocalChecked()); + + global->Set(current_context, String::NewFromUtf8(isolate, "runt_udp_recvfrom").ToLocalChecked(), + Function::New(current_context, v8_runt_udp_recvfrom).ToLocalChecked()); + #ifdef WITH_SSL global->Set(current_context, String::NewFromUtf8(isolate, "runt_socket_enable_ssl").ToLocalChecked(), Function::New(current_context, v8_runt_socket_enable_ssl).ToLocalChecked()); diff --git a/Sources/socket_v8_bindings.h b/Sources/socket_v8_bindings.h index ccf92d3..a0627af 100644 --- a/Sources/socket_v8_bindings.h +++ b/Sources/socket_v8_bindings.h @@ -15,6 +15,10 @@ void v8_runt_socket_close(const v8::FunctionCallbackInfo& args); void v8_runt_socket_set_blocking(const v8::FunctionCallbackInfo& args); void v8_runt_socket_select(const v8::FunctionCallbackInfo& args); void v8_runt_socket_is_connected(const v8::FunctionCallbackInfo& args); +void v8_runt_udp_create(const v8::FunctionCallbackInfo& args); +void v8_runt_udp_bind(const v8::FunctionCallbackInfo& args); +void v8_runt_udp_sendto(const v8::FunctionCallbackInfo& args); +void v8_runt_udp_recvfrom(const v8::FunctionCallbackInfo& args); #ifdef WITH_SSL void v8_runt_socket_enable_ssl(const v8::FunctionCallbackInfo& args); #endif diff --git a/Sources/websocket.cpp b/Sources/websocket.cpp index fcaf918..d057799 100644 --- a/Sources/websocket.cpp +++ b/Sources/websocket.cpp @@ -101,11 +101,19 @@ namespace WebSocketWrapper { void processEvents() { try { + std::vector closed_ids; for (auto& [id, client] : active_websockets) { if (client) { client->processEvents(); + if (client->getReadyState() == CLOSED + && !client->hasPendingEvents()) { + closed_ids.push_back(id); + } } } + for (int id : closed_ids) { + active_websockets.erase(id); + } } catch (...) { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: CRASH in global processEvents"); @@ -120,137 +128,9 @@ namespace WebSocketWrapper { return id; } - class SimpleWebSocketClient { - private: - SOCKET socket_; - std::thread* thread_; - std::atomic connected_{false}; - std::atomic should_stop_{false}; - std::string host_; - int port_; - std::string path_; - - public: - SimpleWebSocketClient(const std::string& url) : socket_(INVALID_SOCKET), thread_(nullptr) { - parseUrl(url); - } - - ~SimpleWebSocketClient() { - disconnect(); - } - - bool connect() { - socket_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); - if (socket_ == INVALID_SOCKET) { - return false; - } - - SocketOptimization::optimizeWebSocket(static_cast(socket_)); - - sockaddr_in addr{}; - addr.sin_family = AF_INET; - addr.sin_port = htons(port_); - inet_pton(AF_INET, host_.c_str(), &addr.sin_addr); - - if (::connect(socket_, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) { - closesocket(socket_); - socket_ = INVALID_SOCKET; - return false; - } - - std::string handshake = - "GET " + path_ + " HTTP/1.1\r\n" - "Host: " + host_ + ":" + std::to_string(port_) + "\r\n" - "Upgrade: websocket\r\n" - "Connection: Upgrade\r\n" - "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" - "Sec-WebSocket-Version: 13\r\n\r\n"; - - if (send(socket_, handshake.c_str(), static_cast(handshake.length()), 0) == SOCKET_ERROR) { - closesocket(socket_); - socket_ = INVALID_SOCKET; - return false; - } - - char buffer[SocketOptimization::SMALL_BUFFER_SIZE]; - int bytes = recv(socket_, buffer, sizeof(buffer) - 1, 0); - if (bytes > 0) { - buffer[bytes] = '\0'; - if (strstr(buffer, "101 Switching Protocols")) { - connected_ = true; - return true; - } - } - - closesocket(socket_); - socket_ = INVALID_SOCKET; - return false; - } - - void disconnect() { - should_stop_ = true; - if (socket_ != INVALID_SOCKET) { - closesocket(socket_); - socket_ = INVALID_SOCKET; - } - if (thread_ && thread_->joinable()) { - thread_->join(); - delete thread_; - thread_ = nullptr; - } - connected_ = false; - } - - bool isConnected() const { - return connected_; - } - - void sendText(const std::string& data) { - if (!connected_ || socket_ == INVALID_SOCKET) return; - - // text frame format FIN + text frame - std::vector frame; - frame.push_back(0x81); - - if (data.length() < 126) { - frame.push_back(0x80 | static_cast(data.length())); // MASK + length - } else { - // TODO: support longer messages - return; - } - - // adds simplified masking key - uint8_t mask[4] = {0x12, 0x34, 0x56, 0x78}; - frame.insert(frame.end(), mask, mask + 4); - - // adds the masked payload - for (size_t i = 0; i < data.length(); i++) { - frame.push_back(data[i] ^ mask[i % 4]); - } - - send(socket_, (char*)frame.data(), static_cast(frame.size()), 0); - } - - private: - void parseUrl(const std::string& url) { - std::regex url_regex(R"(^wss?://([^:/]+)(?::(\d+))?(/.*)?$)"); - std::smatch matches; - - if (std::regex_match(url, matches, url_regex)) { - host_ = matches[1].str(); - port_ = matches[2].matched ? std::stoi(matches[2].str()) : 80; - path_ = matches[3].matched ? matches[3].str() : "/"; - } else { - host_ = "localhost"; - port_ = 80; - path_ = "/"; - } - } - }; - WebSocketClient::WebSocketClient(Isolate* isolate, Global* global_context, const std::string& url) : isolate_(isolate), global_context_(global_context), url_(url), ready_state_(CONNECTING), - is_ssl_(false), buffered_amount_(0), ws_(nullptr), ssl_ws_(nullptr) + is_ssl_(false), buffered_amount_(0), ws_(-1) #ifdef WITH_SSL #ifdef _WIN32 , ssl_cred_handle_(nullptr), ssl_context_handle_(nullptr), ssl_context_initialized_(false), ssl_socket_(-1) @@ -278,21 +158,35 @@ namespace WebSocketWrapper { ready_state_ = CONNECTING; // start connection using thread pool - GlobalThreadPool::getInstance().enqueue([this, host, port, path]() { + connect_future_ = GlobalThreadPool::getInstance().enqueue([this, host, port, path]() { try { + if (ready_state_ == CLOSED) { + return; + } #ifdef DEBUG_NETWORK - kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket connecting to %s:%d%s (SSL: %s)", + kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket connecting to %s:%d%s (SSL: %s)", host.c_str(), port, path.c_str(), is_ssl_ ? "yes" : "no"); #endif - + if (connectToServer(host, port, path)) { - ready_state_ = OPEN; - handleOpen(); - - messageLoop(); + if (ready_state_ == OPEN) { + messageLoop(); + } else { + if (is_ssl_) { + cleanupWebSocketSSL(); + } + if (ws_ >= 0) { + #ifdef _WIN32 + closesocket((SOCKET)ws_); + #else + ::close(ws_); + #endif + ws_ = -1; + } + } } else { - ready_state_ = CLOSED; handleError("Connection failed"); + ready_state_ = CLOSED; } } catch (...) { ready_state_ = CLOSED; @@ -304,9 +198,39 @@ namespace WebSocketWrapper { if (ready_state_ == OPEN) { close(); } + ready_state_ = CLOSED; + if (ws_ >= 0) { + #ifdef _WIN32 + ::shutdown((SOCKET)ws_, SD_BOTH); + #else + ::shutdown(ws_, SHUT_RDWR); + #endif + } + if (connect_future_.valid()) { + connect_future_.wait(); + } + if (ws_ >= 0) { + #ifdef _WIN32 + closesocket((SOCKET)ws_); + #else + ::close(ws_); + #endif + ws_ = -1; + } #ifdef WITH_SSL cleanupWSL(); #endif + if (isolate_ != nullptr) { + v8::Locker locker{isolate_}; + v8::Isolate::Scope isolate_scope(isolate_); + v8::HandleScope handle_scope(isolate_); + on_open_.Reset(); + on_message_.Reset(); + on_error_.Reset(); + on_close_.Reset(); + } + WebSocketEvent ev; + while (event_queue_.try_pop(ev)) {} } bool WebSocketClient::parseUrl(const std::string& url, std::string& host, int& port, std::string& path) { @@ -338,7 +262,7 @@ namespace WebSocketWrapper { return; } - if (ws_ == nullptr) { + if (ws_ < 0) { handleError("WebSocket: No socket available for sending"); return; } @@ -353,7 +277,7 @@ namespace WebSocketWrapper { // create and send WebSocket frame std::vector frame = createWebSocketFrame(data); - int sock = static_cast(reinterpret_cast(ws_)); + int sock = ws_; int sendResult; if (is_ssl_ && isSSLReady()) { @@ -388,7 +312,7 @@ namespace WebSocketWrapper { return; } - if (ws_ == nullptr) { + if (ws_ < 0) { handleError("WebSocket: No socket available for sending"); return; } @@ -403,7 +327,7 @@ namespace WebSocketWrapper { // binary frame with opcode 0x02 std::vector frame = createWebSocketBinaryFrame(data); - int sock = static_cast(reinterpret_cast(ws_)); + int sock = ws_; int sendResult; if (is_ssl_ && isSSLReady()) { @@ -434,8 +358,17 @@ namespace WebSocketWrapper { return; } + if (ready_state_ == OPEN && ws_ >= 0) { + std::vector frame = createCloseFrame(code, reason); + if (is_ssl_ && isSSLReady()) { + webSocketSSLSend(frame.data(), static_cast(frame.size())); + } else { + int sock = ws_; + ::send(sock, reinterpret_cast(frame.data()), static_cast(frame.size()), 0); + } + } + ready_state_ = CLOSING; - ready_state_ = CLOSED; handleClose(code, reason); } @@ -457,16 +390,6 @@ namespace WebSocketWrapper { on_close_.Reset(isolate_, callback); } - // event handlers called from worker thread that queue events safely - void WebSocketClient::handleOpen() { - event_queue_.push(WebSocketEvent(EVENT_OPEN)); - } - - void WebSocketClient::handleMessage(const std::string& message) { - // push directly to lock free queue - event_queue_.push(WebSocketEvent(EVENT_MESSAGE, message)); - } - void WebSocketClient::handleError(const std::string& error) { try { #ifdef DEBUG_NETWORK @@ -479,8 +402,8 @@ namespace WebSocketWrapper { } void WebSocketClient::handleClose(int code, const std::string& reason) { - ready_state_ = CLOSED; event_queue_.push(WebSocketEvent(EVENT_CLOSE, code, reason)); + ready_state_ = CLOSED; } // process events on main thread with lock free queue and batching @@ -928,6 +851,8 @@ namespace WebSocketWrapper { #endif return false; } + + ws_ = sock; sockaddr_in addr{}; addr.sin_family = AF_INET; @@ -945,6 +870,7 @@ namespace WebSocketWrapper { #else close(sock); #endif + ws_ = -1; #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: Failed to resolve host %s", host.c_str()); #endif @@ -961,6 +887,7 @@ namespace WebSocketWrapper { #else close(sock); #endif + ws_ = -1; #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: Failed to connect to %s:%d", host.c_str(), port); #endif @@ -971,7 +898,7 @@ namespace WebSocketWrapper { kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: TCP connected to %s:%d", host.c_str(), port); #endif - ws_ = reinterpret_cast(static_cast(sock)); + ws_ = sock; if (is_ssl_ && port == 443) { if (!initializeWebSocketSSL(sock, host)) { @@ -980,7 +907,7 @@ namespace WebSocketWrapper { #else close(sock); #endif - ws_ = nullptr; + ws_ = -1; return false; } #ifdef DEBUG_NETWORK @@ -995,7 +922,7 @@ namespace WebSocketWrapper { #else close(sock); #endif - ws_ = nullptr; + ws_ = -1; return false; } @@ -1003,9 +930,11 @@ namespace WebSocketWrapper { kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: Handshake completed successfully"); #endif - ready_state_ = OPEN; - fireOpenEvent(); - + if (ready_state_ == CONNECTING) { + ready_state_ = OPEN; + fireOpenEvent(); + } + return true; } catch (const std::exception& e) { @@ -1060,23 +989,35 @@ namespace WebSocketWrapper { } char buffer[SocketOptimization::SMALL_BUFFER_SIZE]; - int bytesReceived; - if (is_ssl_ && isSSLReady()) { - bytesReceived = webSocketSSLReceive(buffer, sizeof(buffer) - 1); - } else { - bytesReceived = recv(sock, buffer, sizeof(buffer) - 1, 0); + std::string response; + for (int attempt = 0; attempt < 64 + && response.find("\r\n\r\n") == std::string::npos; ++attempt) { + int bytesReceived; + if (is_ssl_ && isSSLReady()) { + bytesReceived = webSocketSSLReceive(buffer, sizeof(buffer) - 1); + if (bytesReceived == 0) { + continue; + } + } else { + bytesReceived = recv(sock, buffer, sizeof(buffer) - 1, 0); + } + + if (bytesReceived <= 0) { + #ifdef DEBUG_NETWORK + kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: Failed to receive handshake response"); + #endif + return false; + } + response.append(buffer, bytesReceived); } - - if (bytesReceived <= 0) { + + if (response.find("\r\n\r\n") == std::string::npos) { #ifdef DEBUG_NETWORK - kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: Failed to receive handshake response"); + kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: Handshake response never completed"); #endif return false; } - buffer[bytesReceived] = '\0'; - std::string response(buffer); - #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: Received handshake response"); #endif @@ -1155,8 +1096,8 @@ namespace WebSocketWrapper { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: Connection closed: %d - %s", code, reason.c_str()); #endif - ready_state_ = CLOSED; event_queue_.push(WebSocketEvent(EVENT_CLOSE, code, reason)); + ready_state_ = CLOSED; } // complete frame helper for ring buffer messageLoop @@ -1186,14 +1127,23 @@ namespace WebSocketWrapper { close_reason = std::string(payload.begin() + 2, payload.end()); } } - fireCloseEvent(close_code, close_reason); + if (ready_state_ == OPEN) { + std::vector echoFrame = createCloseFrame(close_code, ""); + if (is_ssl_ && isSSLReady()) { + webSocketSSLSend(echoFrame.data(), static_cast(echoFrame.size())); + } else { + int sock = ws_; + ::send(sock, reinterpret_cast(echoFrame.data()), static_cast(echoFrame.size()), 0); + } + fireCloseEvent(close_code, close_reason); + } } break; case 0x9: // ping frame { std::vector pongFrame = createPongFrame(payload); - int sock = static_cast(reinterpret_cast(ws_)); + int sock = ws_; if (is_ssl_ && isSSLReady()) { webSocketSSLSend(pongFrame.data(), static_cast(pongFrame.size())); } else { @@ -1213,14 +1163,14 @@ namespace WebSocketWrapper { // following uWebSockets void WebSocketClient::messageLoop() { - if (ws_ == nullptr) { + if (ws_ < 0) { #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_ERROR, "WebSocket: No socket available for message loop"); #endif return; } - int sock = static_cast(reinterpret_cast(ws_)); + int sock = ws_; #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: Starting message loop"); #endif @@ -1271,8 +1221,11 @@ namespace WebSocketWrapper { } else { bytesReceived = recv(sock, reinterpret_cast(writePtr), static_cast(contiguousSpace), 0); } - + if (bytesReceived <= 0) { + if (bytesReceived == 0 && is_ssl_ && isSSLReady()) { + continue; + } #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: Connection closed by server"); #endif @@ -1357,6 +1310,10 @@ namespace WebSocketWrapper { } } + if (ready_state_ == OPEN) { + fireCloseEvent(1006, ""); + } + #ifdef DEBUG_NETWORK kinc_log(KINC_LOG_LEVEL_INFO, "WebSocket: Message loop completed"); #endif @@ -1464,7 +1421,7 @@ namespace WebSocketWrapper { case 0x9: { std::vector pongFrame = createPongFrame(payload); - int sock = static_cast(reinterpret_cast(ws_)); + int sock = ws_; int sendResult; if (is_ssl_ && isSSLReady()) { @@ -1964,7 +1921,7 @@ namespace WebSocketWrapper { } } - if (status == SEC_I_CONTINUE_NEEDED && outbuffers[0].cbBuffer > 0 && outbuffers[0].pvBuffer != nullptr) { + if ((status == SEC_I_CONTINUE_NEEDED || status == SEC_E_OK) && outbuffers[0].cbBuffer > 0 && outbuffers[0].pvBuffer != nullptr) { // validate buffer before sending if (outbuffers[0].cbBuffer < 0 || outbuffers[0].cbBuffer > 65536) { kinc_log(KINC_LOG_LEVEL_ERROR, "Invalid WebSocket SSL handshake buffer size: %d bytes (corrupted data)", outbuffers[0].cbBuffer); @@ -2024,6 +1981,10 @@ namespace WebSocketWrapper { } if (status == SEC_E_OK) { + if (!handshake_buffer.empty()) { + ssl_buffer_.insert(ssl_buffer_.end(), handshake_buffer.begin(), + handshake_buffer.end()); + } ssl_context_initialized_ = true; kinc_log(KINC_LOG_LEVEL_INFO, "SChannel WebSocket SSL handshake completed successfully"); return true; @@ -2649,6 +2610,160 @@ void runt_websocket_set_onclose(const FunctionCallbackInfo& args) { } // Native WebSocket class implementation for V8 +static void dispatchWebSocketEvent(const FunctionCallbackInfo& args, const char* type, const char* handler) { + Isolate* isolate = args.GetIsolate(); + HandleScope scope(isolate); + Local context = isolate->GetCurrentContext(); + Local self = args.Data().As(); + + Local event; + if (args.Length() > 0 && args[0]->IsObject()) { + event = args[0].As(); + } else { + event = Object::New(isolate); + } + event->Set(context, String::NewFromUtf8(isolate, "type").ToLocalChecked(), String::NewFromUtf8(isolate, type).ToLocalChecked()); + event->Set(context, String::NewFromUtf8(isolate, "target").ToLocalChecked(), self); + event->Set(context, String::NewFromUtf8(isolate, "currentTarget").ToLocalChecked(), self); + + Local argv[1] = { event }; + Local fn = self->Get(context, String::NewFromUtf8(isolate, handler).ToLocalChecked()).ToLocalChecked(); + if (fn->IsFunction()) { + TryCatch try_catch(isolate); + fn.As()->Call(context, self, 1, argv); + if (try_catch.HasCaught()) { + String::Utf8Value err(isolate, try_catch.Exception()); + kinc_log(KINC_LOG_LEVEL_WARNING, "WebSocket %s handler threw: %s", type, + *err != nullptr ? *err : "unknown"); + } + } + + Local listenersVal = self->Get(context, String::NewFromUtf8(isolate, "_listeners").ToLocalChecked()).ToLocalChecked(); + if (listenersVal->IsObject()) { + Local arr = listenersVal.As()->Get(context, String::NewFromUtf8(isolate, type).ToLocalChecked()).ToLocalChecked(); + if (arr->IsArray()) { + Local listeners = arr.As(); + uint32_t length = listeners->Length(); + std::vector> snapshot; + snapshot.reserve(length); + for (uint32_t i = 0; i < length; ++i) { + Local listener = listeners->Get(context, i).ToLocalChecked(); + if (listener->IsFunction()) { + snapshot.push_back(listener.As()); + } + } + for (Local& listener : snapshot) { + TryCatch try_catch(isolate); + listener->Call(context, self, 1, argv); + if (try_catch.HasCaught()) { + String::Utf8Value err(isolate, try_catch.Exception()); + kinc_log(KINC_LOG_LEVEL_WARNING, "WebSocket %s listener threw: %s", type, + *err != nullptr ? *err : "unknown"); + } + } + } + } +} + +static void WebSocketDispatchOpen(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + HandleScope scope(isolate); + Local context = isolate->GetCurrentContext(); + Local self = args.Data().As(); + + Local idVal = self->Get(context, String::NewFromUtf8(isolate, "_id").ToLocalChecked()).ToLocalChecked(); + if (idVal->IsNumber()) { + int wsId = idVal->Int32Value(context).FromJust(); + auto it = WebSocketWrapper::active_websockets.find(wsId); + if (it != WebSocketWrapper::active_websockets.end()) { + self->Set(context, String::NewFromUtf8(isolate, "protocol").ToLocalChecked(), + String::NewFromUtf8(isolate, it->second->getProtocol().c_str()).ToLocalChecked()); + } + } + dispatchWebSocketEvent(args, "open", "onopen"); +} + +static void WebSocketDispatchMessage(const FunctionCallbackInfo& args) { + dispatchWebSocketEvent(args, "message", "onmessage"); +} + +static void WebSocketDispatchError(const FunctionCallbackInfo& args) { + dispatchWebSocketEvent(args, "error", "onerror"); +} + +static void WebSocketDispatchClose(const FunctionCallbackInfo& args) { + dispatchWebSocketEvent(args, "close", "onclose"); +} + +static void WebSocketReadyStateGetter(Local property, const PropertyCallbackInfo& info) { + Isolate* isolate = info.GetIsolate(); + Local context = isolate->GetCurrentContext(); + int state = WebSocketWrapper::CLOSED; + Local idVal = info.Holder()->Get(context, String::NewFromUtf8(isolate, "_id").ToLocalChecked()).ToLocalChecked(); + if (idVal->IsNumber()) { + auto it = WebSocketWrapper::active_websockets.find(idVal->Int32Value(context).FromJust()); + if (it != WebSocketWrapper::active_websockets.end()) { + state = it->second->getReadyState(); + } + } + info.GetReturnValue().Set(Integer::New(isolate, state)); +} + +static void WebSocketReadyStateSetter(Local property, Local value, const PropertyCallbackInfo& info) {} + +static void WebSocketAddEventListener(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + HandleScope scope(isolate); + Local context = isolate->GetCurrentContext(); + Local self = args.Holder(); + + if (args.Length() < 2 || !args[0]->IsString() || !args[1]->IsFunction()) return; + + Local listenersVal = self->Get(context, String::NewFromUtf8(isolate, "_listeners").ToLocalChecked()).ToLocalChecked(); + if (!listenersVal->IsObject()) return; + Local listeners = listenersVal.As(); + + Local arr = listeners->Get(context, args[0]).ToLocalChecked(); + Local listenerList; + if (arr->IsArray()) { + listenerList = arr.As(); + for (uint32_t i = 0; i < listenerList->Length(); ++i) { + Local existing = listenerList->Get(context, i).ToLocalChecked(); + if (existing->StrictEquals(args[1])) return; + } + } else { + listenerList = Array::New(isolate); + listeners->Set(context, args[0], listenerList); + } + listenerList->Set(context, listenerList->Length(), args[1]); +} + +static void WebSocketRemoveEventListener(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + HandleScope scope(isolate); + Local context = isolate->GetCurrentContext(); + Local self = args.Holder(); + + if (args.Length() < 2 || !args[0]->IsString() || !args[1]->IsFunction()) return; + + Local listenersVal = self->Get(context, String::NewFromUtf8(isolate, "_listeners").ToLocalChecked()).ToLocalChecked(); + if (!listenersVal->IsObject()) return; + + Local arr = listenersVal.As()->Get(context, args[0]).ToLocalChecked(); + if (!arr->IsArray()) return; + Local listenerList = arr.As(); + + for (uint32_t i = 0; i < listenerList->Length(); ++i) { + Local listener = listenerList->Get(context, i).ToLocalChecked(); + if (listener->StrictEquals(args[1])) { + Local splice = listenerList->Get(context, String::NewFromUtf8(isolate, "splice").ToLocalChecked()).ToLocalChecked(); + Local spliceArgs[2] = { Integer::New(isolate, i), Integer::New(isolate, 1) }; + splice.As()->Call(context, listenerList, 2, spliceArgs); + break; + } + } +} + static void WebSocketConstructor(const FunctionCallbackInfo& args) { Isolate* isolate = args.GetIsolate(); HandleScope scope(isolate); @@ -2672,7 +2787,6 @@ static void WebSocketConstructor(const FunctionCallbackInfo& args) { Local instance = args.This(); instance->Set(context, String::NewFromUtf8(isolate, "_id").ToLocalChecked(), Integer::New(isolate, wsId)); instance->Set(context, String::NewFromUtf8(isolate, "url").ToLocalChecked(), args[0]); - instance->Set(context, String::NewFromUtf8(isolate, "readyState").ToLocalChecked(), Integer::New(isolate, 0)); instance->Set(context, String::NewFromUtf8(isolate, "protocol").ToLocalChecked(), String::NewFromUtf8(isolate, "").ToLocalChecked()); instance->Set(context, String::NewFromUtf8(isolate, "extensions").ToLocalChecked(), String::NewFromUtf8(isolate, "").ToLocalChecked()); instance->Set(context, String::NewFromUtf8(isolate, "binaryType").ToLocalChecked(), String::NewFromUtf8(isolate, "blob").ToLocalChecked()); @@ -2681,6 +2795,15 @@ static void WebSocketConstructor(const FunctionCallbackInfo& args) { instance->Set(context, String::NewFromUtf8(isolate, "onmessage").ToLocalChecked(), Null(isolate)); instance->Set(context, String::NewFromUtf8(isolate, "onerror").ToLocalChecked(), Null(isolate)); instance->Set(context, String::NewFromUtf8(isolate, "onclose").ToLocalChecked(), Null(isolate)); + instance->Set(context, String::NewFromUtf8(isolate, "_listeners").ToLocalChecked(), Object::New(isolate)); + + auto it = WebSocketWrapper::active_websockets.find(wsId); + if (it != WebSocketWrapper::active_websockets.end()) { + it->second->setOnOpen(Function::New(context, WebSocketDispatchOpen, instance).ToLocalChecked()); + it->second->setOnMessage(Function::New(context, WebSocketDispatchMessage, instance).ToLocalChecked()); + it->second->setOnError(Function::New(context, WebSocketDispatchError, instance).ToLocalChecked()); + it->second->setOnClose(Function::New(context, WebSocketDispatchClose, instance).ToLocalChecked()); + } args.GetReturnValue().Set(instance); } @@ -2745,8 +2868,6 @@ static void WebSocketClose(const FunctionCallbackInfo& args) { if (readyState == 2 || readyState == 3) return; - self->Set(context, String::NewFromUtf8(isolate, "readyState").ToLocalChecked(), Integer::New(isolate, 2)); - Local idVal = self->Get(context, String::NewFromUtf8(isolate, "_id").ToLocalChecked()).ToLocalChecked(); int wsId = idVal->Int32Value(context).FromJust(); @@ -2883,10 +3004,15 @@ void bind_websocket_class(Isolate* isolate, const Global& context) { Local wsTpl = FunctionTemplate::New(isolate, WebSocketConstructor); wsTpl->SetClassName(String::NewFromUtf8(isolate, "WebSocket").ToLocalChecked()); wsTpl->InstanceTemplate()->SetInternalFieldCount(1); + wsTpl->InstanceTemplate()->SetNativeDataProperty( + String::NewFromUtf8(isolate, "readyState").ToLocalChecked(), + WebSocketReadyStateGetter, WebSocketReadyStateSetter); Local proto = wsTpl->PrototypeTemplate(); proto->Set(isolate, "send", FunctionTemplate::New(isolate, WebSocketSend)); proto->Set(isolate, "close", FunctionTemplate::New(isolate, WebSocketClose)); + proto->Set(isolate, "addEventListener", FunctionTemplate::New(isolate, WebSocketAddEventListener)); + proto->Set(isolate, "removeEventListener", FunctionTemplate::New(isolate, WebSocketRemoveEventListener)); Local wsFunc = wsTpl->GetFunction(current_context).ToLocalChecked(); wsFunc->Set(current_context, String::NewFromUtf8(isolate, "CONNECTING").ToLocalChecked(), Integer::New(isolate, 0)); @@ -2895,6 +3021,11 @@ void bind_websocket_class(Isolate* isolate, const Global& context) { wsFunc->Set(current_context, String::NewFromUtf8(isolate, "CLOSED").ToLocalChecked(), Integer::New(isolate, 3)); global->Set(current_context, String::NewFromUtf8(isolate, "WebSocket").ToLocalChecked(), wsFunc); + + Local windowVal = global->Get(current_context, String::NewFromUtf8(isolate, "window").ToLocalChecked()).ToLocalChecked(); + if (windowVal->IsObject()) { + windowVal.As()->Set(current_context, String::NewFromUtf8(isolate, "WebSocket").ToLocalChecked(), wsFunc); + } } #endif diff --git a/Sources/websocket.h b/Sources/websocket.h index e810c5c..332660c 100644 --- a/Sources/websocket.h +++ b/Sources/websocket.h @@ -11,6 +11,7 @@ #include #include #include +#include #include #include @@ -68,6 +69,7 @@ namespace WebSocketWrapper { void processEvents(); ReadyState getReadyState() const { return ready_state_; } + bool hasPendingEvents() const { return !event_queue_.empty(); } const std::string& getUrl() const { return url_; } const std::string& getProtocol() const { return protocol_; } int getBufferedAmount() const { return buffered_amount_; } @@ -83,7 +85,7 @@ namespace WebSocketWrapper { v8::Global* global_context_; std::string url_; std::string protocol_; - ReadyState ready_state_; + std::atomic ready_state_; bool is_ssl_; int buffered_amount_; @@ -94,9 +96,9 @@ namespace WebSocketWrapper { v8::Global on_close_; LockFreeQueue event_queue_; - - void* ws_; - void* ssl_ws_; + std::future connect_future_; + + int ws_; #ifdef WITH_SSL #ifdef _WIN32 @@ -131,8 +133,6 @@ namespace WebSocketWrapper { std::string base64Encode(const std::string& data); // called from worker thread - void handleOpen(); - void handleMessage(const std::string& message); void handleError(const std::string& error); void handleClose(int code, const std::string& reason); diff --git a/kfile.js b/kfile.js index 97e3080..abf4db1 100644 --- a/kfile.js +++ b/kfile.js @@ -13,6 +13,7 @@ let flags = { with_webview: true, with_uws: false, with_ssl: true, + with_rtc: true, debug_network: false }; @@ -122,6 +123,30 @@ if (flags.with_viewport && (platform === Platform.Windows || platform === Platfo project.addFile('Sources/viewport_server.h'); project.addFile('Sources/viewport_server.cpp'); } + +if (flags.with_rtc && (platform === Platform.Windows || platform === Platform.Linux || platform === Platform.OSX)) { + project.addDefine('WITH_RTC'); + project.addDefine('RTC_STATIC'); + project.addFile('Sources/rtc_bridge.h'); + project.addFile('Sources/rtc_bridge.cpp'); + project.addFile('Sources/rtc_v8_bindings.h'); + project.addFile('Sources/rtc_v8_bindings.cpp'); + project.addIncludeDir('libdatachannel/include'); + if (platform === Platform.Windows) { + project.addLib(root + '/libdatachannel/lib/datachannel-static'); + project.addLib(root + '/libdatachannel/lib/juice-static'); + project.addLib(root + '/libdatachannel/lib/usrsctp'); + project.addLib(root + '/libdatachannel/lib/MbedTLS'); + project.addLib('bcrypt'); + } + else { + project.addLib(root + '/libdatachannel/lib/libdatachannel-static.a'); + project.addLib(root + '/libdatachannel/lib/libjuice-static.a'); + project.addLib(root + '/libdatachannel/lib/libusrsctp.a'); + project.addLib(root + '/libdatachannel/lib/libMbedTLS.a'); + } +} + if (flags.with_networking && platform !== Platform.Windows) { project.addLib('ssl'); project.addLib('crypto'); @@ -142,7 +167,7 @@ if (platform === Platform.Windows) { project.addLib(libdir + 'v8_monolith'); if (flags.with_webview) { project.addIncludeDir('webview2/include'); - project.addLib(root + '/webview2/lib/x64/WebView2Loader'); + project.addLib(root + '/webview2/lib/x64/WebView2LoaderStatic'); } if (!flags.release) { project.addDefine('_HAS_ITERATOR_DEBUGGING=0');