From bc214ed9224ef15e5cb73d9d0fe73900e5e4bfe5 Mon Sep 17 00:00:00 2001 From: drslebedev Date: Fri, 7 Aug 2026 12:51:34 +0200 Subject: [PATCH] RestClient: run Emscripten fetches on a dedicated worker Emscripten delivers fetch callbacks on the thread that issued the request, after that thread returns to the JS event loop. ClientContext worked around this by starting a std::thread for every command and immediately joining it, blocking its poller. Direct RestClient users such as OpenDigitizer had no workaround, so fetch callbacks could run on the browser main thread. Give each Emscripten RestClient a persistent worker and route requests, long polling, unsubscribe and cleanup through it. Resume a 504 long-poll timeout at the same index instead of ending the subscription. Fetch completion and subscription callbacks now run on the REST worker, so consumers must synchronise shared state. Each active client uses one worker, which is reclaimed asynchronously during shutdown. Remove the per-request thread workaround from ClientContext. The native RestClient implementation is unchanged. Add an Emscripten integration test covering subscriptions, unsubscribe, shutdown, callback threading and worker reclamation. The Node test requires the xhr2 npm package. Browser tests are registered when emrun and a matching browser are available; OPENCMW_REQUIRE_BROWSER_TESTS requires at least one browser test. Signed-off-by: drslebedev --- src/client/include/ClientContext.hpp | 11 +- src/client/include/RestClientEmscripten.hpp | 740 +++++++++++------- src/client/test/CMakeLists.txt | 53 ++ .../EmscriptenRestClientTest.cpp | 222 ++++++ .../test/emscripten_rest_client/node_setup.js | 15 + src/client/test/emscripten_rest_client/run.py | 133 ++++ 6 files changed, 878 insertions(+), 296 deletions(-) create mode 100644 src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp create mode 100644 src/client/test/emscripten_rest_client/node_setup.js create mode 100644 src/client/test/emscripten_rest_client/run.py diff --git a/src/client/include/ClientContext.hpp b/src/client/include/ClientContext.hpp index 0e917a35..9edc7651 100644 --- a/src/client/include/ClientContext.hpp +++ b/src/client/include/ClientContext.hpp @@ -88,15 +88,7 @@ class ClientContext { return false; } auto &c = getClientCtx(cmd.topic); -#ifdef EMSCRIPTEN - // this is necessary for fetches to actually be called, as the new thread will start/init/end and then go into js runtime to fetch - std::thread ql{ [&c, cmd]() { -#endif - c.request(cmd); -#ifdef EMSCRIPTEN - } }; - ql.join(); -#endif + c.request(std::move(cmd)); return false; }); } @@ -112,6 +104,7 @@ class ClientContext { void queueCommand(mdp::Command cmd, const URI &endpoint, std::function &&callback = {}, IoBuffer &&data = IoBuffer{}) { bool published = _commandRingBuffer->tryPublishEvent([&endpoint, &cmd, cb = std::move(callback), d = std::move(data)](Command &&ev, long /*seq*/) mutable { + ev = Command{}; ev.command = cmd; ev.callback = std::move(cb); ev.topic = FWD(endpoint); diff --git a/src/client/include/RestClientEmscripten.hpp b/src/client/include/RestClientEmscripten.hpp index 24bbcc49..2712c472 100644 --- a/src/client/include/RestClientEmscripten.hpp +++ b/src/client/include/RestClientEmscripten.hpp @@ -2,11 +2,35 @@ #define OPENCMW_CPP_RESTCLIENT_EMSCRIPTEN_HPP #include +#include #include - +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include +#include +#include +#include +#include +#include #include -#include +#include +#include +#include +#include +#include +#include #include #include @@ -19,241 +43,483 @@ namespace opencmw::client { namespace detail { -/*** - * Get the final URL of a possibly redirected HTTP fetch call. - * Uses Javascript to return the the url as a string. - */ -#pragma GCC diagnostic push -#pragma GCC diagnostic ignored "-Wdollar-in-identifier-extension" -static std::string getFinalURL(std::uint32_t id) { - auto finalURLChar = static_cast(EM_ASM_PTR({ - var fetch = Fetch.xhrs.get($0); - if (fetch) { - var finalURL = fetch.responseURL; - var lengthBytes = lengthBytesUTF8(finalURL) + 1; - var stringOnWasmHeap = _malloc(lengthBytes); - stringToUTF8(finalURL, stringOnWasmHeap, lengthBytes); - return stringOnWasmHeap; - } - return 0; }, id)); - if (finalURLChar == nullptr) { +struct RestWorkerState; + +inline std::string_view responseBody(const emscripten_fetch_t *fetch) noexcept { + if (fetch->data == nullptr || fetch->numBytes == 0) { return {}; } - std::string finalURL{ finalURLChar, strlen(finalURLChar) }; - EM_ASM({ _free($0) }, finalURLChar); - return finalURL; + const auto maximum = static_cast(std::numeric_limits::max()); + return { fetch->data, static_cast(std::min(fetch->numBytes, maximum)) }; } -#pragma GCC diagnostic pop - -struct pointer_equals { - using is_transparent = void; - template - bool operator()(const Left &left, const Right &right) const { - return std::to_address(left) == std::to_address(right); +inline std::optional parseLongPollingIndex(std::string_view responseUrl) noexcept { + if (responseUrl.empty()) { + return std::nullopt; } + try { + const auto params = URI<>(std::string{ responseUrl }).queryParamMap(); + const auto entry = params.find("LongPollingIdx"); + if (entry == params.end() || !entry->second || entry->second->empty()) { + return std::nullopt; + } + const std::string &value = *entry->second; + std::uint64_t index{}; + const auto [end, error] = std::from_chars(value.data(), value.data() + value.size(), index); + return error == std::errc{} && end == value.data() + value.size() ? std::optional{ index } : std::nullopt; + } catch (...) { + return std::nullopt; + } +} + +struct SubscriptionState { + Command command{}; + std::optional lastDeliveredIndex{}; + std::optional activeFetchId{}; +}; + +struct ActiveFetch { + RestWorkerState *owner{ nullptr }; + std::uint64_t id{}; + std::optional subscriptionId{}; // absent for GET/SET + std::optional command{}; // present for GET/SET + std::string body{}; // must outlive the fetch + emscripten_fetch_t *fetch{ nullptr }; + bool closing{ false }; }; -struct pointer_hash { - using is_transparent = void; +struct RestWorkerState { + std::atomic _acceptWork{ true }; + std::shared_ptr _selfKeepAlive{}; + + MIME::MimeType _mimeType; + + std::unordered_map _subscriptions{}; + std::unordered_map> _activeFetches{}; + std::uint64_t _nextSubscriptionId{ 1 }; + std::uint64_t _nextFetchId{ 1 }; + + explicit RestWorkerState(MIME::MimeType mimeType) + : _mimeType(mimeType) {} - template - std::size_t operator()(const Pointer &ptr) const { - const auto *raw = std::to_address(ptr); - return std::hash{}(raw); + void dispatchCommand(Command &&cmd) noexcept { + if (!_acceptWork.load(std::memory_order_acquire)) { + return; + } + Command failure; + try { + failure.topic = cmd.topic; + failure.clientRequestID = cmd.clientRequestID; + failure.callback = cmd.callback; + + switch (cmd.command) { + case mdp::Command::Get: + case mdp::Command::Set: startGetOrSet(std::move(cmd)); return; + case mdp::Command::Subscribe: startSubscription(std::move(cmd)); return; + case mdp::Command::Unsubscribe: stopSubscription(cmd); return; + default: + reportFailure(failure, "command type is undefined"); + return; + } + } catch (const std::exception &e) { + reportFailure(failure, e.what()); + } catch (...) { + reportFailure(failure, "failed to start command"); + } } -}; -auto checkedStringViewSize = [](auto numBytes) { - if (numBytes > std::numeric_limits::max()) { - throw std::out_of_range(std::format("We received more data than we can handle {}", numBytes)); + void startSubscription(Command &&cmd) { + const std::uint64_t id = _nextSubscriptionId++; + _subscriptions.emplace(id, SubscriptionState{ .command = std::move(cmd) }); + startNextLongPoll(id, std::nullopt); } - return static_cast(numBytes); -}; -std::array getPreferredContentTypeHeader(const URI &uri, auto _mimeType) { - auto mimeType = std::string(_mimeType.typeName()); - if (const auto acceptHeader = uri.queryParamMap().find("contentType"); acceptHeader != uri.queryParamMap().end() && acceptHeader->second) { - mimeType = acceptHeader->second->c_str(); + void stopSubscription(const Command &cmd) { + const auto entry = std::ranges::find_if(_subscriptions, + [&](const auto &pair) { return pair.second.command.topic == cmd.topic; }); + if (entry == _subscriptions.end()) { + return; + } + const std::optional outstandingFetchId = entry->second.activeFetchId; + _subscriptions.erase(entry); + if (outstandingFetchId.has_value()) { + closeFetch(*outstandingFetchId); + } } - return { "accept", mimeType, "content-type", mimeType }; -} -struct FetchPayload { - Command command; + void startNextLongPoll(std::uint64_t subscriptionId, std::optional index) noexcept { + try { + if (!_acceptWork.load(std::memory_order_acquire)) { + return; + } + const auto entry = _subscriptions.find(subscriptionId); + if (entry == _subscriptions.end()) { + return; + } + const std::string longPollingIndex = index.has_value() ? std::to_string(*index) : "Next"; + + auto activeFetch = std::make_unique(); + activeFetch->owner = this; + activeFetch->id = _nextFetchId++; + activeFetch->subscriptionId = subscriptionId; + entry->second.activeFetchId = activeFetch->id; + startFetch(std::move(activeFetch), URI::UriFactory(entry->second.command.topic).addQueryParameter("LongPollingIdx", longPollingIndex).build()); + } catch (const std::exception &e) { + endSubscription(subscriptionId, nullptr, 500, {}, e.what()); + } catch (...) { + endSubscription(subscriptionId, nullptr, 500, {}, "failed to start long-poll request"); + } + } - explicit FetchPayload(Command &&_command) - : command(std::move(_command)) {} + void startGetOrSet(Command &&cmd) { + const URI uri = cmd.topic; - FetchPayload(const FetchPayload &other) = delete; + auto activeFetch = std::make_unique(); + activeFetch->owner = this; + activeFetch->id = _nextFetchId++; + if (cmd.command == mdp::Command::Set) { + activeFetch->body = cmd.data.asString(); + } + activeFetch->command = std::move(cmd); - FetchPayload(FetchPayload &&other) noexcept = default; + startFetch(std::move(activeFetch), uri); + } - FetchPayload &operator=(const FetchPayload &other) = delete; + void startFetch(std::unique_ptr activeFetch, const URI &uri) { + std::string contentType{ _mimeType.typeName() }; + const auto &query = uri.queryParamMap(); + if (const auto entry = query.find("contentType"); entry != query.end() && entry->second) { + contentType = *entry->second; + } + const std::array headers{ "accept", contentType.c_str(), "content-type", contentType.c_str(), nullptr }; - FetchPayload &operator=(FetchPayload &&other) noexcept = default; + const std::string_view method = activeFetch->command.has_value() && activeFetch->command->command == mdp::Command::Set ? "POST" : "GET"; - void returnMdpMessage(unsigned short status, std::string_view body, std::string_view errorMsgExt = "") noexcept { - if (!command.callback) { + emscripten_fetch_attr_t attr; + emscripten_fetch_attr_init(&attr); + method.copy(attr.requestMethod, method.size()); + attr.requestMethod[method.size()] = '\0'; + attr.attributes = EMSCRIPTEN_FETCH_LOAD_TO_MEMORY; + attr.requestHeaders = headers.data(); + attr.onsuccess = &RestWorkerState::onFetchSuccess; + attr.onerror = &RestWorkerState::onFetchError; + attr.userData = activeFetch.get(); + if (!activeFetch->body.empty()) { + attr.requestData = activeFetch->body.data(); + attr.requestDataSize = activeFetch->body.size(); + } + + const std::uint64_t fetchId = activeFetch->id; + _activeFetches.emplace(fetchId, std::move(activeFetch)); + + emscripten_fetch_t *fetch = emscripten_fetch(&attr, uri.str().c_str()); + + if (fetch == nullptr) { + const auto entry = _activeFetches.find(fetchId); + if (entry == _activeFetches.end()) { + return; + } + const std::optional subscriptionId = entry->second->subscriptionId; + const std::optional command = std::move(entry->second->command); + _activeFetches.erase(entry); + + if (subscriptionId.has_value()) { + endSubscription(*subscriptionId, nullptr, 500, {}, "emscripten_fetch() returned null"); + } else if (command.has_value()) { + reportFailure(*command, "emscripten_fetch() returned null"); + } return; } - const bool msgOK = status >= 200 && status < 400; + + if (const auto entry = _activeFetches.find(fetchId); entry != _activeFetches.end() && !entry->second->closing) { + entry->second->fetch = fetch; + } + } + + static void onFetchSuccess(emscripten_fetch_t *fetch) noexcept { completeFetch(fetch, true); } + static void onFetchError(emscripten_fetch_t *fetch) noexcept { completeFetch(fetch, false); } + + static void completeFetch(emscripten_fetch_t *fetch, bool succeeded) noexcept { + auto *activeFetch = static_cast(fetch->userData); + if (activeFetch == nullptr || activeFetch->owner == nullptr || activeFetch->closing) { + return; + } + RestWorkerState *owner = activeFetch->owner; + const std::uint64_t fetchId = activeFetch->id; + const auto subscriptionId = activeFetch->subscriptionId; try { - command.callback(mdp::Message{ - .id = 0, - .arrivalTime = std::chrono::system_clock::now(), - .protocolName = command.topic.scheme().value(), - .command = mdp::Command::Final, - .clientRequestID = command.clientRequestID, - .topic = command.topic, - .data = msgOK ? IoBuffer(body.data(), body.size()) : IoBuffer(), - .error = msgOK ? std::string(errorMsgExt) : std::format("{} - {}{}{}", status, errorMsgExt, body.empty() ? "" : ":", body), - .rbac = IoBuffer() }); + // This view must be consumed before the handler closes the fetch. + std::optional fetchError; + if (!succeeded) { + fetchError = fetch->statusText[0] != '\0' ? std::string_view{ fetch->statusText } : std::string_view{ "fetch failed" }; + } + if (subscriptionId.has_value()) { + owner->handleSubscriptionCompletion(fetchId, *subscriptionId, fetch, std::move(fetchError)); + } else { + owner->handleGetOrSetCompletion(fetchId, fetch, std::move(fetchError)); + } } catch (const std::exception &e) { - std::cerr - << std::format("caught exception '{}' in FetchPayload::returnMdpMessage(cmd={}, {}: {})", e.what(), command.topic, status, - body) - << std::endl; + owner->discardFetch(fetchId, subscriptionId, fetch, e.what()); + std::println(std::cerr, "RestClientEmscripten: fetch callback failed: {}", e.what()); } catch (...) { - std::cerr - << std::format("caught unknown exception in FetchPayload::returnMdpMessage(cmd={}, {}: {})", command.topic, status, body) - << std::endl; + owner->discardFetch(fetchId, subscriptionId, fetch, "fetch callback failed"); + std::println(std::cerr, "RestClientEmscripten: fetch callback failed"); } } - void onsuccess(unsigned short status, std::string_view data) { - returnMdpMessage(status, data); - } + void handleSubscriptionCompletion(std::uint64_t fetchId, std::uint64_t subscriptionId, emscripten_fetch_t *fetch, std::optional fetchError) { + if (!_acceptWork.load(std::memory_order_acquire)) { + closeFetch(fetchId, fetch); + return; + } + const auto entry = _subscriptions.find(subscriptionId); + if (entry == _subscriptions.end()) { + closeFetch(fetchId, fetch); + return; + } + SubscriptionState &state = entry->second; - void onerror(unsigned short status, std::string_view error, std::string_view data) { - returnMdpMessage(status, data, error); - } -}; + const unsigned short status = fetch->status; + const std::string_view body = responseBody(fetch); + const auto index = parseLongPollingIndex(fetch->responseUrl != nullptr ? std::string_view{ fetch->responseUrl } : std::string_view{}); -static std::unordered_set, detail::pointer_hash, detail::pointer_equals> fetchPayloads; + // Server timeout on long-poll, resend the same request. + if (status == 504) { + if (!index.has_value()) { + endSubscription(subscriptionId, fetch, status, body, "missing or unparsable LongPollingIdx in the response URL"); + return; + } + closeFetch(fetchId, fetch); + startNextLongPoll(subscriptionId, *index); + return; + } -struct SubscriptionPayload; -static std::unordered_set, detail::pointer_hash, detail::pointer_equals> subscriptionPayloads; + if (fetchError.has_value()) { + endSubscription(subscriptionId, fetch, status, body, *fetchError); + return; + } -struct SubscriptionPayload : FetchPayload { - bool _live = true; - MIME::MimeType _mimeType; - std::size_t _update = 0; + if (!index.has_value()) { + endSubscription(subscriptionId, fetch, status, body, "missing or unparsable LongPollingIdx in the response URL"); + return; + } - static constexpr std::size_t kParallelLongPollingRequests = 1; // increasing this value could reduce latency but needs some more robust error handling for unexpected updates - std::vector _requestedIndexes; + if (state.lastDeliveredIndex.has_value() && *index <= *state.lastDeliveredIndex) { + const std::uint64_t expected = *state.lastDeliveredIndex + 1; + closeFetch(fetchId, fetch); + startNextLongPoll(subscriptionId, expected); + return; + } - SubscriptionPayload(Command &&_command, MIME::MimeType mimeType) - : FetchPayload(std::move(_command)), _mimeType(std::move(mimeType)) {} + std::string skippedWarning; + if (state.lastDeliveredIndex.has_value() && *index - *state.lastDeliveredIndex > 1) { + skippedWarning = std::format("Warning: skipped {} samples", *index - *state.lastDeliveredIndex - 1); + } - SubscriptionPayload(const SubscriptionPayload &other) = delete; + const mdp::Message message = buildMessage(state.command, status, body, skippedWarning); + state.lastDeliveredIndex = *index; - SubscriptionPayload(SubscriptionPayload &&other) noexcept = default; + closeFetch(fetchId, fetch); + invokeGuarded(state.command.callback, message); + startNextLongPoll(subscriptionId, *index + 1); + } - SubscriptionPayload &operator=(const SubscriptionPayload &other) = delete; + void handleGetOrSetCompletion(std::uint64_t fetchId, emscripten_fetch_t *fetch, std::optional fetchError) { + if (!_acceptWork.load(std::memory_order_acquire)) { + closeFetch(fetchId, fetch); + return; + } + const auto entry = _activeFetches.find(fetchId); + if (entry == _activeFetches.end() || !entry->second->command.has_value()) { + closeFetch(fetchId, fetch); + return; + } + const unsigned short status = fetch->status; + const Command command = std::move(*entry->second->command); - SubscriptionPayload &operator=(SubscriptionPayload &&other) noexcept = default; + std::optional message; + try { + message = buildMessage(command, status, responseBody(fetch), fetchError.has_value() ? std::string_view{ *fetchError } : std::string_view{}); + } catch (const std::exception &e) { + std::println(std::cerr, "RestClientEmscripten: could not build the GET/SET response: {}", e.what()); + } - void sendFollowUpRequestsFor(std::uint64_t longPollingIdx) { - auto it = std::ranges::find(_requestedIndexes, longPollingIdx); - if (it != _requestedIndexes.end()) { - _requestedIndexes.erase(it); + closeFetch(fetchId, fetch); + if (message.has_value()) { + invokeGuarded(command.callback, *message); } - for (std::uint64_t i = longPollingIdx + 1; i <= longPollingIdx + kParallelLongPollingRequests; ++i) { - if (std::ranges::find(_requestedIndexes, i) == _requestedIndexes.end()) { - _requestedIndexes.push_back(i); - request(std::to_string(i)); - } + } + + void discardFetch(std::uint64_t fetchId, std::optional subscriptionId, emscripten_fetch_t *callbackFetch, std::string_view error) noexcept { + if (subscriptionId.has_value() && _subscriptions.contains(*subscriptionId)) { + endSubscription(*subscriptionId, callbackFetch, 500, {}, error); + } else { + closeFetch(fetchId, callbackFetch); } } - void request(std::string longPollingIndex) { - auto uri = opencmw::URI::UriFactory(command.topic).addQueryParameter("LongPollingIdx", longPollingIndex).build(); - auto preferredHeader = detail::getPreferredContentTypeHeader(command.topic, _mimeType); - std::array preferredHeaderEmscripten; - std::transform(preferredHeader.cbegin(), preferredHeader.cend(), preferredHeaderEmscripten.begin(), - [](const auto &str) { return str.c_str(); }); - preferredHeaderEmscripten[preferredHeaderEmscripten.size() - 1] = nullptr; + void endSubscription(std::uint64_t subscriptionId, emscripten_fetch_t *callbackFetch, unsigned short status, std::string_view body, std::string_view error) noexcept { + const auto entry = _subscriptions.find(subscriptionId); + if (entry == _subscriptions.end()) { + return; + } + const Command command = std::move(entry->second.command); + const std::optional outstandingFetchId = entry->second.activeFetchId; + _subscriptions.erase(entry); - emscripten_fetch_attr_t attr{}; + std::optional message; + try { + message = buildMessage(command, status, body, error); + } catch (const std::exception &e) { + std::println(std::cerr, "RestClientEmscripten: could not report '{}': {}", error, e.what()); + } catch (...) { + std::println(std::cerr, "RestClientEmscripten: could not report '{}'", error); + } - emscripten_fetch_attr_init(&attr); + if (outstandingFetchId.has_value()) { + closeFetch(*outstandingFetchId, callbackFetch); + } + if (message.has_value()) { + invokeGuarded(command.callback, *message); + } + } + + void closeFetch(std::uint64_t fetchId, emscripten_fetch_t *callbackFetch = nullptr) noexcept { + const auto entry = _activeFetches.find(fetchId); + if (entry == _activeFetches.end() || entry->second->closing) { + return; + } + entry->second->closing = true; + if (emscripten_fetch_t *fetch = callbackFetch != nullptr ? callbackFetch : entry->second->fetch; fetch != nullptr) { + (void) emscripten_fetch_close(fetch); + } + _activeFetches.erase(fetchId); + } - strcpy(attr.requestMethod, "GET"); + void cleanup() noexcept { + while (!_activeFetches.empty()) { + closeFetch(_activeFetches.begin()->first); + } + _subscriptions.clear(); + _selfKeepAlive.reset(); + emscripten_runtime_keepalive_pop(); + } - attr.userData = this; - static auto getPayloadIt = [](emscripten_fetch_t *fetch) { - auto *rawPayload = fetch->userData; - auto it = detail::subscriptionPayloads.find(rawPayload); - if (it == detail::subscriptionPayloads.end()) { - std::print("RestClientEmscripten::payloadError: url: {}, bytes: {}\n", fetch->url, fetch->numBytes); - throw std::format("Unknown payload for a resulting subscription"); - } - return it; + void reportFailure(const Command &command, std::string_view error) noexcept { + if (!command.callback) { + std::println(std::cerr, "RestClientEmscripten: {}", error); + return; + } + try { + invokeGuarded(command.callback, buildMessage(command, 500, {}, error)); + } catch (const std::exception &e) { + std::println(std::cerr, "RestClientEmscripten: could not report '{}': {}", error, e.what()); + } catch (...) { + std::println(std::cerr, "RestClientEmscripten: could not report '{}'", error); + } + } + + void invokeGuarded(const std::function &callback, const mdp::Message &message) noexcept { + if (!callback || !_acceptWork.load(std::memory_order_acquire)) { + return; + } + try { + callback(message); + } catch (const std::exception &e) { + std::println(std::cerr, "RestClientEmscripten: callback threw '{}'", e.what()); + } catch (...) { + std::println(std::cerr, "RestClientEmscripten: callback threw"); + } + } + + static mdp::Message buildMessage(const Command &command, unsigned short status, std::string_view body, std::string_view error) { + const bool ok = status >= 200 && status < 400; + return mdp::Message{ + .id = 0, + .arrivalTime = std::chrono::system_clock::now(), + .protocolName = command.topic.scheme().value_or(""), + .command = mdp::Command::Final, + .clientRequestID = command.clientRequestID, + .topic = command.topic, + .data = ok ? IoBuffer(body.data(), body.size()) : IoBuffer(), + .error = ok ? std::string(error) : std::format("{} - {}{}{}", status, error, body.empty() ? "" : ":", body), + .rbac = IoBuffer() }; + } +}; - attr.attributes = EMSCRIPTEN_FETCH_LOAD_TO_MEMORY; - attr.requestHeaders = preferredHeaderEmscripten.data(); - attr.onsuccess = [](emscripten_fetch_t *fetch) { - auto payloadIt = getPayloadIt(fetch); - auto &payload = *payloadIt; - std::uint64_t longPollingIdx = 0; - if (payload->_live) { - std::string finalURL = getFinalURL(fetch->id); - std::string longPollingIdxString = opencmw::URI<>(finalURL).queryParamMap().at("LongPollingIdx").value_or("0"); - - char *end = nullptr; - longPollingIdx = strtoull(longPollingIdxString.data(), &end, 10); - if (end != longPollingIdxString.data() + longPollingIdxString.size()) { - std::println(std::cerr, "RestClientEmscripten::payloadError: url: {}, bytes: {}\n", fetch->url, fetch->numBytes); +class FetchWorker { + std::shared_ptr _queue{ std::make_shared() }; + std::shared_ptr _state; + std::mutex _enqueueMutex{}; // keeps cleanup behind accepted commands + pthread_t _worker{}; + +public: + explicit FetchWorker(MIME::MimeType mimeType) + : _state(std::make_shared(mimeType)) { + if (_queue->queue == nullptr) { + throw std::runtime_error("RestClient: proxying queue allocation failed"); + } + + // Keep the detached pthread runtime alive to process proxied work. + std::thread worker{ [] { emscripten_runtime_keepalive_push(); } }; + _worker = worker.native_handle(); + worker.detach(); + _state->_selfKeepAlive = _state; + } + + ~FetchWorker() { stop(); } + + FetchWorker(const FetchWorker &) = delete; + FetchWorker &operator=(const FetchWorker &) = delete; + FetchWorker(FetchWorker &&) = delete; + FetchWorker &operator=(FetchWorker &&) = delete; + + void submit(Command &&cmd) { + std::shared_ptr pendingCommand; + try { + { + std::lock_guard lock(_enqueueMutex); + if (!_state->_acceptWork.load(std::memory_order_acquire)) { return; } - const long indexDiff = static_cast(longPollingIdx) - static_cast(payload->_update + 1); - if (payload->_update != 0 && indexDiff != 0) { - std::print("received unexpected update: {}, expected {}\n", longPollingIdx, payload->_update + 1); + pendingCommand = std::make_shared(std::move(cmd)); + if (_queue->proxyAsync(_worker, [state = _state, command = pendingCommand]() mutable { state->dispatchCommand(std::move(*command)); })) { + return; } - payload->onsuccess(fetch->status, std::string_view(fetch->data, detail::checkedStringViewSize(fetch->numBytes)), indexDiff); - emscripten_fetch_close(fetch); - - payload->_update = longPollingIdx; - payload->sendFollowUpRequestsFor(longPollingIdx); - } else { - detail::subscriptionPayloads.erase(payloadIt); } - }; - attr.onerror = [](emscripten_fetch_t *fetch) { - auto payloadIt = getPayloadIt(fetch); - auto &payload = *payloadIt; - payload->onerror(fetch->status, std::string_view(fetch->data, detail::checkedStringViewSize(fetch->numBytes)), fetch->statusText); - emscripten_fetch_close(fetch); - }; - emscripten_fetch(&attr, uri.str().data()); - } - - void onsuccess(unsigned short status, std::string_view data, long idxDifference = 0) { - std::string skippedWarning; - if (idxDifference != 0) { - skippedWarning = std::format("Warning: skipped {} samples", idxDifference); + _state->reportFailure(*pendingCommand, "request was not queued on the REST worker"); + } catch (const std::exception &e) { + _state->reportFailure(pendingCommand ? *pendingCommand : cmd, e.what()); + } catch (...) { + _state->reportFailure(pendingCommand ? *pendingCommand : cmd, "could not queue request on the REST worker"); } - returnMdpMessage(status, data, skippedWarning); } - void onerror(unsigned short status, std::string_view error, std::string_view data) { - returnMdpMessage(status, data, error); + void stop() noexcept { + if (!_state->_acceptWork.exchange(false, std::memory_order_acq_rel)) { + return; + } + try { + std::lock_guard lock(_enqueueMutex); + if (_queue->proxyAsync(_worker, [state = _state, queue = _queue] { state->cleanup(); })) { + return; + } + } catch (...) { // locking failed, or proxyAsync could not allocate the task + } + std::fputs("RestClientEmscripten: could not queue REST worker cleanup; leaving the worker alive\n", stderr); } }; } // namespace detail class RestClient : public ClientBase { - std::string _name; - MIME::MimeType _mimeType = MIME::BINARY; - std::atomic _run = true; - std::string _caCertificate; + std::string _name; + MIME::MimeType _mimeType; + std::string _caCertificate; + detail::FetchWorker _worker; public: - static bool CHECK_CERTIFICATES; - /** * Initialises a basic RestClient * @@ -264,128 +530,28 @@ class RestClient : public ClientBase { * @param initArgs */ template + requires(!(std::same_as, RestClient> || ...)) explicit(false) RestClient(Args... initArgs) : _name(detail::find_argument_value([] { return "RestClient"; }, initArgs...)) - , _mimeType(detail::find_argument_value([] { return MIME::BINARY; }, initArgs...)) { - } - ~RestClient() { RestClient::stop(); } - - void stop() override {} - - std::vector protocols() noexcept override { return { "http", "https" }; } - - [[nodiscard]] std::string name() const noexcept { return _name; } - // [[nodiscard]] ThreadPoolType threadPool() const noexcept { return _thread_pool; } - [[nodiscard]] MIME::MimeType defaultMimeType() const noexcept { return _mimeType; } - [[nodiscard]] std::string clientCertificate() const noexcept { return _caCertificate; } - - void request(Command cmd) override { - switch (cmd.command) { - case mdp::Command::Get: - case mdp::Command::Set: - executeCommand(std::move(cmd)); - return; - case mdp::Command::Subscribe: - startSubscription(std::move(cmd)); - return; - case mdp::Command::Unsubscribe: // deregister existing subscription URI is key - stopSubscription(std::move(cmd)); - return; - default: - throw std::invalid_argument("command type is undefined"); - } - } - -private: - void executeCommand(Command &&cmd) const { - auto preferredHeader = detail::getPreferredContentTypeHeader(cmd.topic, _mimeType); - std::array preferredHeaderEmscripten; - std::transform(preferredHeader.cbegin(), preferredHeader.cend(), preferredHeaderEmscripten.begin(), - [](const auto &str) { return str.c_str(); }); - preferredHeaderEmscripten[preferredHeaderEmscripten.size() - 1] = nullptr; - - emscripten_fetch_attr_t attr; - emscripten_fetch_attr_init(&attr); - - auto payload = std::make_unique(std::move(cmd)); - attr.userData = payload.get(); - - if (payload->command.command == opencmw::mdp::Command::Set) { - strcpy(attr.requestMethod, "POST"); - auto body = payload->command.data.asString(); - attr.requestData = body.data(); - attr.requestDataSize = body.size(); - } else { - strcpy(attr.requestMethod, "GET"); - } + , _mimeType(detail::find_argument_value([] { return MIME::BINARY; }, initArgs...)) + , _worker(_mimeType) {} - static auto getPayload = [](emscripten_fetch_t *fetch) { - auto *rawPayload = fetch->userData; - auto it = detail::fetchPayloads.find(rawPayload); - if (it == detail::fetchPayloads.end()) { - throw std::format("Unknown payload for a resulting fetch call"); - } - auto extracted_node = detail::fetchPayloads.extract(it); - return std::move(extracted_node.value()); - }; + ~RestClient() override = default; - attr.attributes = EMSCRIPTEN_FETCH_LOAD_TO_MEMORY; - attr.requestHeaders = preferredHeaderEmscripten.data(); - attr.onsuccess = [](emscripten_fetch_t *fetch) { - getPayload(fetch)->onsuccess(fetch->status, std::string_view(fetch->data, detail::checkedStringViewSize(fetch->numBytes))); - emscripten_fetch_close(fetch); - }; - attr.onerror = [](emscripten_fetch_t *fetch) { - getPayload(fetch)->onerror(fetch->status, std::string_view(fetch->data, detail::checkedStringViewSize(fetch->numBytes)), fetch->statusText); - emscripten_fetch_close(fetch); - }; + RestClient(const RestClient &) = delete; + RestClient &operator=(const RestClient &) = delete; + RestClient(RestClient &&) = delete; + RestClient &operator=(RestClient &&) = delete; - // TODO: Pass the payload as POST body: emscripten_fetch(&attr, uri.relativeRef()->data()); + void stop() override { _worker.stop(); } - emscripten_fetch(&attr, payload->command.topic.str().data()); - detail::fetchPayloads.insert(std::move(payload)); - } + std::vector protocols() noexcept override { return { "http", "https" }; } - void startSubscription(Command &&cmd) { - auto payload = std::make_unique(std::move(cmd), _mimeType); - auto rawPayload = payload.get(); - detail::subscriptionPayloads.insert(std::move(payload)); - std::print("starting subscription: {}, existing subscriptions: {}, from main thread: \n", cmd.topic.str(), detail::subscriptionPayloads.size(), emscripten_is_main_runtime_thread()); - if (emscripten_is_main_runtime_thread()) { - try { - rawPayload->request("Next"); - } catch (std::runtime_error &e) { - rawPayload->onerror(500, e.what(), ""); - } catch (...) { - rawPayload->onerror(500, "failed to set up subscription", ""); - } - } else { - emscripten_async_run_in_main_runtime_thread(EM_FUNC_SIG_IP, +[](void *data) { - auto subPayload = reinterpret_cast(data); - try { - subPayload->request("Next"); - } catch (std::runtime_error &e) { - subPayload->onerror(500, e.what(), ""); - } catch (...) { - subPayload->onerror(500, "failed to set up subscription", ""); - } - return 0; }, rawPayload); - } - } - - void stopSubscription(Command &&cmd) { - auto payloadIt = std::ranges::find_if(detail::subscriptionPayloads, - [&](const auto &ptr) { - return ptr->command.topic == cmd.topic; - }); - if (payloadIt == detail::subscriptionPayloads.end()) { - return; - } - std::print("stopping subscription: {}, existing subscriptions: {}\n", cmd.topic.str(), detail::subscriptionPayloads.size()); + [[nodiscard]] std::string name() const noexcept { return _name; } + [[nodiscard]] MIME::MimeType defaultMimeType() const noexcept { return _mimeType; } + [[nodiscard]] std::string clientCertificate() const noexcept { return _caCertificate; } - auto &payload = *payloadIt; - payload->_live = false; - } + void request(Command cmd) override { _worker.submit(std::move(cmd)); } }; } // namespace opencmw::client diff --git a/src/client/test/CMakeLists.txt b/src/client/test/CMakeLists.txt index ea4c7727..1ee9d5b5 100644 --- a/src/client/test/CMakeLists.txt +++ b/src/client/test/CMakeLists.txt @@ -75,6 +75,59 @@ target_include_directories(rest_client_only_tests PRIVATE ${CMAKE_SOURCE_DIR}) # catch_discover_tests(rest_client_only_tests) if(EMSCRIPTEN) + set(EMSCRIPTEN_REST_CLIENT_TEST_DIR ${CMAKE_CURRENT_SOURCE_DIR}/emscripten_rest_client) + add_executable(emscripten_rest_client_tests ${EMSCRIPTEN_REST_CLIENT_TEST_DIR}/EmscriptenRestClientTest.cpp) + set_target_properties( + emscripten_rest_client_tests + PROPERTIES SUFFIX ".html" + LINK_DEPENDS ${EMSCRIPTEN_REST_CLIENT_TEST_DIR}/node_setup.js) + target_link_libraries( + emscripten_rest_client_tests + PUBLIC opencmw_project_warnings + opencmw_project_options + client) + target_link_options( + emscripten_rest_client_tests + PRIVATE + --emrun + --pre-js=${EMSCRIPTEN_REST_CLIENT_TEST_DIR}/node_setup.js) + + find_package(Python3 REQUIRED COMPONENTS Interpreter) + set(REST_CLIENT_TEST_RUNNER ${EMSCRIPTEN_REST_CLIENT_TEST_DIR}/run.py) + + add_test( + NAME emscripten_rest_client_tests_node + COMMAND ${Python3_EXECUTABLE} ${REST_CLIENT_TEST_RUNNER} --mode node --node ${CMAKE_CROSSCOMPILING_EMULATOR} + --binary $/emscripten_rest_client_tests.js) + set_tests_properties(emscripten_rest_client_tests_node PROPERTIES TIMEOUT 60) + + option(OPENCMW_REQUIRE_BROWSER_TESTS "Fail configuration when the Emscripten browser tests cannot be registered" OFF) + get_filename_component(EMSCRIPTEN_TOOLS_DIR ${CMAKE_CXX_COMPILER} DIRECTORY) + find_program(EMRUN_EXECUTABLE NAMES emrun HINTS ${EMSCRIPTEN_TOOLS_DIR}) + find_program(CHROME_EXECUTABLE NAMES google-chrome google-chrome-stable chromium chromium-browser) + find_program(FIREFOX_EXECUTABLE NAMES firefox firefox-esr HINTS /snap/firefox/current/usr/lib/firefox) + if(CHROME_EXECUTABLE AND EMRUN_EXECUTABLE) + add_test( + NAME emscripten_rest_client_tests_chrome + COMMAND ${Python3_EXECUTABLE} ${REST_CLIENT_TEST_RUNNER} --mode browser --browser-family chromium --browser + ${CHROME_EXECUTABLE} --emrun ${EMRUN_EXECUTABLE} --binary $) + set_tests_properties(emscripten_rest_client_tests_chrome PROPERTIES TIMEOUT 120) + endif() + if(FIREFOX_EXECUTABLE AND EMRUN_EXECUTABLE) + add_test( + NAME emscripten_rest_client_tests_firefox + COMMAND ${Python3_EXECUTABLE} ${REST_CLIENT_TEST_RUNNER} --mode browser --browser-family firefox --browser + ${FIREFOX_EXECUTABLE} --emrun ${EMRUN_EXECUTABLE} --binary $) + set_tests_properties(emscripten_rest_client_tests_firefox PROPERTIES TIMEOUT 120) + endif() + if(NOT EMRUN_EXECUTABLE OR (NOT CHROME_EXECUTABLE AND NOT FIREFOX_EXECUTABLE)) + if(OPENCMW_REQUIRE_BROWSER_TESTS) + message(FATAL_ERROR "OPENCMW_REQUIRE_BROWSER_TESTS is set but emrun or a supported browser was not found") + else() + message(STATUS "emrun or a supported browser was not found - browser REST client tests not registered") + endif() + endif() + add_executable(emscripten_client_tests EmscriptenClientTests.cpp) target_link_libraries( emscripten_client_tests diff --git a/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp new file mode 100644 index 00000000..aaeb0696 --- /dev/null +++ b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp @@ -0,0 +1,222 @@ +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +using namespace opencmw; +using namespace opencmw::client; + +namespace { + +constexpr int kStreamACount = 5; + +struct TestState { + int failures{}; + int initialRunningWorkers{}; + int initialUnusedWorkers{}; + std::optional client; + std::atomic_int messagesA{}; + std::atomic_int messagesB{}; + std::atomic_bool sawMainThread{}; + std::atomic_bool sawWorkerThread{}; + std::string receivedA; + std::string receivedB; + int callbackCountAtCleanup{}; + std::chrono::steady_clock::time_point deadline; +}; + +TestState &testState(void *data) { + return *static_cast(data); +} + +void check(TestState &state, bool condition, std::string_view failure, std::source_location location = std::source_location::current()) { + if (!condition) { + std::println("{}:{}: FAIL: {}", location.file_name(), location.line(), failure); + ++state.failures; + } +} + +// PThread worker counts are Emscripten internals used only for this leak check. +int runningWorkerCount() noexcept { + return EM_ASM_INT({ return PThread.runningWorkers.length; }); +} + +int unusedWorkerCount() noexcept { + return EM_ASM_INT({ return PThread.unusedWorkers.length; }); +} + +bool workerPoolRestored(const TestState &state) noexcept { + return runningWorkerCount() == state.initialRunningWorkers && unusedWorkerCount() == state.initialUnusedWorkers; +} + +std::string testPayload(int index) { + std::string expected = std::format("{}:", index); + for (int i = 0; i < 100; ++i) { + std::format_to(std::back_inserter(expected), "{}", i); + } + return expected; +} + +void recordCallbackThread(TestState &state) { + if (emscripten_is_main_runtime_thread()) { + state.sawMainThread.store(true, std::memory_order_relaxed); + } else { + state.sawWorkerThread.store(true, std::memory_order_relaxed); + } +} + +void reportAndExit(const TestState &state) { + std::println("=== {} ({} failure{}) ===", state.failures == 0 ? "PASSED" : "FAILED", state.failures, state.failures == 1 ? "" : "s"); + emscripten_force_exit(state.failures == 0 ? 0 : 1); +} + +void finishTest(void *data) { + constexpr int kStreamAFirst = 7; + constexpr int kStreamBFirst = 5; + + auto &state = testState(data); + const int messagesA = state.messagesA.load(std::memory_order_acquire); + const int messagesB = state.messagesB.load(std::memory_order_acquire); + const int callbacks = messagesA + messagesB; + + check(state, state.callbackCountAtCleanup == callbacks, std::format("callback count changed after cleanup ({} to {})", state.callbackCountAtCleanup, callbacks)); + check(state, workerPoolRestored(state), "worker count changed after cleanup"); + check(state, messagesA == kStreamACount, std::format("stream A delivered {} messages, expected {}", messagesA, kStreamACount)); + check(state, messagesB == 1, std::format("stream B delivered {} messages, expected 1", messagesB)); + check(state, !state.sawMainThread.load(std::memory_order_relaxed), "a callback ran on the browser main thread"); + check(state, state.sawWorkerThread.load(std::memory_order_relaxed), "no callback ran on the REST worker"); + + std::string expectedA; + for (int index = kStreamAFirst; index < kStreamAFirst + kStreamACount; ++index) { + expectedA += testPayload(index); + } + const std::string expectedB = testPayload(kStreamBFirst); + check(state, state.receivedA == expectedA, std::format("stream A payload differs ({} bytes, expected {})", state.receivedA.size(), expectedA.size())); + check(state, state.receivedB == expectedB, std::format("stream B payload differs ({} bytes, expected {})", state.receivedB.size(), expectedB.size())); + + reportAndExit(state); +} + +void waitForCleanup(void *); + +void waitForDelivery(void *data) { + // Must exceed PROBE_DELAY_SECONDS in run.py. + constexpr auto kUnsubscribeSettle = std::chrono::milliseconds{ 1500 }; + + auto &state = testState(data); + const auto now = std::chrono::steady_clock::now(); + if (state.messagesA.load(std::memory_order_acquire) >= kStreamACount && state.messagesB.load(std::memory_order_acquire) >= 1) { + emscripten_cancel_main_loop(); + // Allow the delayed sixth response to expose a failed unsubscribe. + emscripten_set_timeout([](void *callbackData) { + constexpr auto kCleanupTimeout = std::chrono::seconds{ 10 }; + + auto &callbackState = testState(callbackData); + const auto begin = std::chrono::steady_clock::now(); + callbackState.client->stop(); + const auto elapsed = std::chrono::duration_cast(std::chrono::steady_clock::now() - begin); + + check(callbackState, elapsed < std::chrono::seconds{ 5 }, std::format("stop() took {} ms with a long poll open", elapsed.count())); + callbackState.client.reset(); + + callbackState.deadline = std::chrono::steady_clock::now() + kCleanupTimeout; + waitForCleanup(callbackData); + }, + kUnsubscribeSettle.count(), data); + return; + } + if (now > state.deadline) { + check(state, false, "timed out waiting for subscription messages"); + reportAndExit(state); + } +} + +void waitForCleanup(void *data) { + constexpr auto kPollInterval = std::chrono::milliseconds{ 20 }; + constexpr auto kStabilityWindow = std::chrono::milliseconds{ 100 }; + + auto &state = testState(data); + const auto now = std::chrono::steady_clock::now(); + if (workerPoolRestored(state)) { + state.callbackCountAtCleanup = state.messagesA.load(std::memory_order_acquire) + state.messagesB.load(std::memory_order_acquire); + emscripten_set_timeout(&finishTest, kStabilityWindow.count(), data); + return; + } + if (now > state.deadline) { + check(state, false, std::format("worker pool not restored (running {}/{}, unused {}/{})", runningWorkerCount(), state.initialRunningWorkers, unusedWorkerCount(), state.initialUnusedWorkers)); + reportAndExit(state); + return; + } + emscripten_set_timeout(&waitForCleanup, kPollInterval.count(), data); +} + +} // namespace + +int main(int argc, char **argv) { + constexpr std::string_view portFlag = "--port="; + constexpr auto kDeliveryTimeout = std::chrono::seconds{ 15 }; + // Main-loop callbacks retain this state for the process lifetime. + static TestState state; + + int port = 0; + for (int i = 1; i < argc; ++i) { + if (const std::string_view arg{ argv[i] }; arg.starts_with(portFlag)) { + port = std::atoi(arg.data() + portFlag.size()); + } + } + if (port == 0) { + std::println("no server port: start this through emscripten_rest_client/run.py"); + return 2; + } + + const URI topicA(std::format("http://127.0.0.1:{}/streamA", port)); + const URI topicB(std::format("http://127.0.0.1:{}/streamB", port)); + + std::println("Emscripten RestClient integration test (server on port {})", port); + + state.initialRunningWorkers = runningWorkerCount(); + state.initialUnusedWorkers = unusedWorkerCount(); + state.deadline = std::chrono::steady_clock::now() + kDeliveryTimeout; + + state.client.emplace(); + + Command subscribeA; + subscribeA.command = mdp::Command::Subscribe; + subscribeA.topic = topicA; + subscribeA.callback = [test = &state, topicA](const mdp::Message &message) { + recordCallbackThread(*test); + test->receivedA += message.data.asString(); + if (test->messagesA.load(std::memory_order_relaxed) == kStreamACount - 1) { + Command unsubscribe; + unsubscribe.command = mdp::Command::Unsubscribe; + unsubscribe.topic = topicA; + test->client->request(std::move(unsubscribe)); + } + test->messagesA.fetch_add(1, std::memory_order_release); + }; + state.client->request(std::move(subscribeA)); + + Command subscribeB; + subscribeB.command = mdp::Command::Subscribe; + subscribeB.topic = topicB; + subscribeB.callback = [test = &state](const mdp::Message &message) { + recordCallbackThread(*test); + test->receivedB += message.data.asString(); + test->messagesB.fetch_add(1, std::memory_order_release); + }; + state.client->request(std::move(subscribeB)); + + emscripten_set_main_loop_arg(&waitForDelivery, &state, 0, EM_TRUE); +} diff --git a/src/client/test/emscripten_rest_client/node_setup.js b/src/client/test/emscripten_rest_client/node_setup.js new file mode 100644 index 00000000..f000cd94 --- /dev/null +++ b/src/client/test/emscripten_rest_client/node_setup.js @@ -0,0 +1,15 @@ +// Node has no XMLHttpRequest; use xhr2 for Emscripten's Fetch API. +if (typeof XMLHttpRequest === 'undefined') { + XMLHttpRequest = require('xhr2'); + + // xhr2 does not report abort completion, so notify Emscripten to release the fetch keepalive. + const abort = XMLHttpRequest.prototype.abort; + XMLHttpRequest.prototype.abort = function () { + const inFlight = this.readyState > 0 && this.readyState < XMLHttpRequest.DONE; + abort.call(this); + if (inFlight) { + this.readyState = XMLHttpRequest.DONE; + this.onreadystatechange?.(); + } + }; +} diff --git a/src/client/test/emscripten_rest_client/run.py b/src/client/test/emscripten_rest_client/run.py new file mode 100644 index 00000000..c6423edd --- /dev/null +++ b/src/client/test/emscripten_rest_client/run.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 + +import argparse +import http.server +import subprocess +import threading +from urllib.parse import parse_qs, urlparse + +# The delayed sixth stream-A response makes a failed unsubscribe observable. +STREAMS = { + "/streamA": (7, 8, 9, 10, 11, 12), + "/streamB": (5,), +} +PROBE_INDEX = ("/streamA", 12) + +def payload(index): + return "{}:{}".format(index, "".join(str(i) for i in range(100))).encode() + +HOLD_SECONDS = 30.0 +PROBE_DELAY_SECONDS = 0.5 + +stopping = threading.Event() + +class Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_OPTIONS(self): + self.send_response(204) + self._common_headers() + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "accept, content-type") + self.send_header("Content-Length", "0") + self.end_headers() + + def do_GET(self): + parsed = urlparse(self.path) + indices = STREAMS.get(parsed.path) + if indices is None: + self._respond(404, b"unknown stream") + return + + index = parse_qs(parsed.query).get("LongPollingIdx", [""])[0] + if index == "Next": + self._redirect(parsed.path, min(indices)) + return + if not index.isdigit(): + self._respond(400, b"malformed LongPollingIdx") + return + + if int(index) not in indices: + stopping.wait(HOLD_SECONDS) + self._respond(504, b"") + return + if (parsed.path, int(index)) == PROBE_INDEX: + stopping.wait(PROBE_DELAY_SECONDS) + self._respond(200, payload(int(index))) + + def _common_headers(self): + self.send_header("Access-Control-Allow-Origin", "*") + + def _redirect(self, path, index): + # Absolute, because xhr2 does not resolve a relative Location against the request URL. + location = "http://{}{}?LongPollingIdx={}".format(self.headers["Host"], path, index) + try: + self.send_response(302) + self._common_headers() + self.send_header("Location", location) + self.send_header("Content-Length", "0") + self.end_headers() + except (BrokenPipeError, ConnectionResetError): + pass + + def _respond(self, code, body): + try: + self.send_response(code) + self._common_headers() + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except (BrokenPipeError, ConnectionResetError): + pass # Expected when the client aborts the long poll. + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--mode", choices=("node", "browser"), required=True) + parser.add_argument("--binary", required=True, help="generated .js (Node) or .html (browser) test program") + parser.add_argument("--node", help="node executable (node mode)") + parser.add_argument("--browser", help="browser executable (browser mode)") + parser.add_argument("--browser-family", choices=("chromium", "firefox"), help="browser family (browser mode)") + parser.add_argument("--emrun", help="emrun executable (browser mode)") + args = parser.parse_args() + if args.mode == "node" and not args.node: + parser.error("--node is required in node mode") + if args.mode == "browser" and not all((args.browser, args.browser_family, args.emrun)): + parser.error("--browser, --browser-family, and --emrun are required in browser mode") + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + threading.Thread(target=server.serve_forever, daemon=True).start() + + if args.mode == "node": + command = [args.node, args.binary, "--port={}".format(port)] + else: + if args.browser_family == "chromium": + browser_args = "--headless=new --no-sandbox --disable-gpu --disable-dev-shm-usage" + else: + browser_args = "--headless" + command = [ + args.emrun, + "--browser", args.browser, + "--browser-args={}".format(browser_args), + "--port", "0", + "--kill-exit", + "--silence-timeout", "60", + ] + if args.browser_family == "firefox": + command.append("--safe-firefox-profile") + command.extend([args.binary, "--", "--port={}".format(port)]) + + try: + return subprocess.call(command) + finally: + stopping.set() + server.shutdown() + + +if __name__ == "__main__": + raise SystemExit(main())