diff --git a/src/client/include/ClientContext.hpp b/src/client/include/ClientContext.hpp index 0e917a35..36a1790d 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; }); } diff --git a/src/client/include/RestClientEmscripten.hpp b/src/client/include/RestClientEmscripten.hpp index 24bbcc49..b0aabc0c 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,492 @@ 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 }; + // node's XHR shim fires onreadystatechange from abort(), so emscripten_fetch_close() re-enters + // completeFetch(); without this guard that recurses until the stack overflows. + bool closing{ false }; }; -struct pointer_hash { - using is_transparent = void; +struct RestWorkerState { + std::atomic _acceptWork{ true }; + std::shared_ptr _shutdownLifetime{}; // deliberate self-reference, released by cleanup() + + 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: + std::println(std::cerr, "RestClientEmscripten: unsupported command {}", static_cast(cmd.command)); + 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); + } + + 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 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 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"); + } } - return { "accept", mimeType, "content-type", mimeType }; -} -struct FetchPayload { - Command command; + void startGetOrSet(Command &&cmd) { + const URI uri = cmd.topic; - explicit FetchPayload(Command &&_command) - : command(std::move(_command)) {} + 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(const FetchPayload &other) = delete; + startFetch(std::move(activeFetch), uri); + } - FetchPayload(FetchPayload &&other) noexcept = default; + 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=(const FetchPayload &other) = delete; + const std::string_view method = activeFetch->command.has_value() && activeFetch->command->command == mdp::Command::Set ? "POST" : "GET"; - FetchPayload &operator=(FetchPayload &&other) noexcept = default; + 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(); + } - void returnMdpMessage(unsigned short status, std::string_view body, std::string_view errorMsgExt = "") noexcept { - if (!command.callback) { + const std::uint64_t fetchId = activeFetch->id; + // Register first because emscripten_fetch() may invoke a completion callback before returning. + _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; + } + + // A synchronous completion may already have erased this state. + 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; } - const bool msgOK = status >= 200 && status < 400; + 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()); + } - 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); // by key: the nested callback may have rehashed the map + } - strcpy(attr.requestMethod, "GET"); + void cleanup() noexcept { + while (!_activeFetches.empty()) { + closeFetch(_activeFetches.begin()->first); + } + _subscriptions.clear(); + _shutdownLifetime.reset(); // safe: the proxied task running this still holds a reference + 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 { + // Tasks capture both, so neither dies before the work queued on it. Dropping the last queue + // reference from a running task is safe: em_task_queue_destroy defers a notified queue. + std::shared_ptr _queue{ std::make_shared() }; + std::shared_ptr _state; + std::mutex _admissionMutex{}; // orders stop()'s cleanup after every admitted command + 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->_shutdownLifetime = _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(_admissionMutex); + 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); + // Retain the command until the queue has accepted ownership of the task. + pendingCommand = std::make_shared(std::move(cmd)); + if (_queue->proxyAsync(_worker, [state = _state, queue = _queue, 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); + // Closes work admission synchronously; the worker tears itself down asynchronously. + void stop() noexcept { + // Cleared before the lock, so admission ends even if locking throws. + if (!_state->_acceptWork.exchange(false, std::memory_order_acq_rel)) { + return; + } + try { + std::lock_guard lock(_admissionMutex); + if (_queue->proxyAsync(_worker, [state = _state, queue = _queue] { state->cleanup(); })) { + return; + } + } catch (...) { // locking failed, or proxyAsync could not allocate the task + } + // Deliberate leak: _shutdownLifetime keeps the state alive for outstanding fetch callbacks. + // fputs, not println: this path may be out of memory, so reporting must not allocate. + 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 +539,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..013a9870 100644 --- a/src/client/test/CMakeLists.txt +++ b/src/client/test/CMakeLists.txt @@ -75,6 +75,58 @@ 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") + 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) + + # CMAKE_CROSSCOMPILING_EMULATOR is reused because the workflow points it at the system node, + # emsdk's being too old for threading. + 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..3f4c164a --- /dev/null +++ b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp @@ -0,0 +1,216 @@ +// Exercises the real RestClient through ClientContext against the HTTP server in run.py. +// The scenario matches OpenDigitizer: subscriptions are submitted by the context poller, +// callbacks run on the REST worker, and shutdown happens on the browser main thread. +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +using namespace opencmw; +using namespace opencmw::client; + +namespace { + +int g_failures = 0; + +void check(bool condition, std::string_view what) { + std::println(" {} {}", condition ? "ok: " : "FAIL:", what); + if (!condition) { + ++g_failures; + } +} + +int runningWorkerCount() noexcept { + return EM_ASM_INT({ return PThread.runningWorkers.length; }); +} +int unusedWorkerCount() noexcept { + return EM_ASM_INT({ return PThread.unusedWorkers.length; }); +} + +int g_initialRunningWorkers{}; +int g_initialUnusedWorkers{}; + +bool workerPoolRestored() noexcept { + return runningWorkerCount() == g_initialRunningWorkers && unusedWorkerCount() == g_initialUnusedWorkers; +} + +constexpr int kStreamAFirst = 7; +constexpr int kStreamACount = 5; +constexpr int kStreamBFirst = 5; +constexpr auto kPollInterval = std::chrono::milliseconds{ 20 }; +constexpr auto kDeliveryTimeout = std::chrono::seconds{ 15 }; +constexpr auto kUnsubscribeSettle = std::chrono::milliseconds{ 1500 }; +constexpr auto kCleanupTimeout = std::chrono::seconds{ 10 }; +constexpr auto kStabilityWindow = std::chrono::milliseconds{ 100 }; + +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; +} + +std::string g_topicA; +std::string g_topicB; +std::optional g_context; + +std::atomic_int g_messagesA{ 0 }; +std::atomic_int g_messagesB{ 0 }; +std::atomic_bool g_sawMainThread{ false }; +std::atomic_bool g_sawWorkerThread{ false }; + +std::string g_receivedA; +std::string g_receivedB; +int g_callbackCountAtCleanup{}; + +int callbackCount() noexcept { + return g_messagesA.load(std::memory_order_acquire) + g_messagesB.load(std::memory_order_acquire); +} + +void recordCallbackThread() { + if (emscripten_is_main_runtime_thread()) { + g_sawMainThread.store(true, std::memory_order_relaxed); + } else { + g_sawWorkerThread.store(true, std::memory_order_relaxed); + } +} + +void startScenario() { + std::vector> clients; + clients.emplace_back(std::make_unique()); + g_context.emplace(std::move(clients)); + + g_context->subscribe(URI(g_topicA), [](const mdp::Message &message) { + recordCallbackThread(); + g_receivedA += message.data.asString(); + if (g_messagesA.fetch_add(1, std::memory_order_release) == kStreamACount - 1) { + // This is how OpenDigitizer stops a subscription from its callback. + g_context->unsubscribe(URI(g_topicA)); + } + }); + g_context->subscribe(URI(g_topicB), [](const mdp::Message &message) { + recordCallbackThread(); + g_receivedB += message.data.asString(); + g_messagesB.fetch_add(1, std::memory_order_release); + }); +} + +void reportAndExit() { + std::println("=== {} ({} failure{}) ===", g_failures == 0 ? "PASSED" : "FAILED", g_failures, g_failures == 1 ? "" : "s"); + emscripten_force_exit(g_failures == 0 ? 0 : 1); +} + +void failAndExit(std::string_view failure) { + check(false, failure); + reportAndExit(); +} + +void finishTest(void *) { + check(g_callbackCountAtCleanup == callbackCount(), "callback counts remained stable after cleanup"); + check(workerPoolRestored(), "the REST and ClientContext workers returned to Emscripten's pool"); + check(g_messagesA.load(std::memory_order_acquire) == kStreamACount, std::format("the unsubscribed chain delivered exactly {} messages", kStreamACount)); + check(g_messagesB.load(std::memory_order_acquire) == 1, "the held subscription delivered exactly one message"); + check(!g_sawMainThread.load(std::memory_order_relaxed), "no callback ran on the browser main thread"); + check(g_sawWorkerThread.load(std::memory_order_relaxed), "callbacks ran on the REST worker"); + + std::string expectedA; + for (int index = kStreamAFirst; index < kStreamAFirst + kStreamACount; ++index) { + expectedA += testPayload(index); + } + check(g_receivedA == expectedA, std::format("the whole chain arrived intact and in order ({} bytes)", g_receivedA.size())); + check(g_receivedB == testPayload(kStreamBFirst), std::format("the held subscription payload arrived intact ({} bytes)", g_receivedB.size())); + + reportAndExit(); +} + +std::chrono::steady_clock::time_point g_deadline; + +void waitForCleanup(void *); + +void beginShutdown(void *) { + const auto begin = std::chrono::steady_clock::now(); + g_context->stop(); + const auto elapsed = std::chrono::duration_cast(std::chrono::steady_clock::now() - begin); + + check(elapsed < std::chrono::seconds{ 5 }, std::format("stop() returned promptly with a long-poll held open ({} ms)", elapsed.count())); + g_context.reset(); + + g_deadline = std::chrono::steady_clock::now() + kCleanupTimeout; + waitForCleanup(nullptr); +} + +void failCleanup() { + failAndExit(std::format("workers were not reclaimed (running {}, expected {}; unused {}, expected {})", runningWorkerCount(), g_initialRunningWorkers, unusedWorkerCount(), g_initialUnusedWorkers)); +} + +void waitForDelivery(void *) { + const auto now = std::chrono::steady_clock::now(); + if (g_messagesA.load(std::memory_order_acquire) >= kStreamACount && g_messagesB.load(std::memory_order_acquire) >= 1) { + // Outlasts streamA's delayed probe index, so an ignored unsubscribe fails the count + // check below rather than going unnoticed (see PROBE_DELAY_SECONDS in run.py). + emscripten_set_timeout(&beginShutdown, kUnsubscribeSettle.count(), nullptr); + return; + } + if (now > g_deadline) { + failAndExit("both subscriptions reported before the deadline"); + return; + } + emscripten_set_timeout(&waitForDelivery, kPollInterval.count(), nullptr); +} + +void waitForCleanup(void *) { + const auto now = std::chrono::steady_clock::now(); + if (workerPoolRestored()) { + g_callbackCountAtCleanup = callbackCount(); + emscripten_set_timeout(&finishTest, kStabilityWindow.count(), nullptr); + return; + } + if (now > g_deadline) { + failCleanup(); + return; + } + emscripten_set_timeout(&waitForCleanup, kPollInterval.count(), nullptr); +} + +} // namespace + +int main(int argc, char **argv) { + constexpr std::string_view portFlag = "--port="; + + 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; + } + + g_topicA = std::format("http://127.0.0.1:{}/streamA", port); + g_topicB = std::format("http://127.0.0.1:{}/streamB", port); + + std::println("=== Emscripten RestClient integration test (server on port {}) ===", port); + + g_initialRunningWorkers = runningWorkerCount(); + g_initialUnusedWorkers = unusedWorkerCount(); + g_deadline = std::chrono::steady_clock::now() + kDeliveryTimeout; + startScenario(); + emscripten_set_timeout(&waitForDelivery, kPollInterval.count(), nullptr); + return 0; +} 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..80a509d6 --- /dev/null +++ b/src/client/test/emscripten_rest_client/node_setup.js @@ -0,0 +1,14 @@ +// emscripten's Fetch.js calls `new XMLHttpRequest()`, which `node` does not provide. +if (typeof XMLHttpRequest === 'undefined') { + XMLHttpRequest = require('xhr2'); + + 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..1ba983ad --- /dev/null +++ b/src/client/test/emscripten_rest_client/run.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 + +import argparse +import http.server +import subprocess +import threading +from urllib.parse import parse_qs, urlparse + +# Indices each stream has; an index past the last is held open, like a long-poll with no new data. +# streamA holds one index more than the test consumes, so an ignored unsubscribe shows up as a +# sixth message instead of looking like the end of the stream. +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 +# Long enough for the unsubscribe to abort the in-flight probe, short enough to answer it if not. +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 # the client aborted the fetch, which is one of the things under test + + +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() + + 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", + ] + command.extend([args.binary, "--", "--port={}".format(port)]) + + try: + return subprocess.call(command) + finally: + stopping.set() + server.shutdown() + + +if __name__ == "__main__": + raise SystemExit(main())