Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix issues with HTTP subscriptions #356

Merged
merged 4 commits into from
Nov 7, 2024
Merged

Fix issues with HTTP subscriptions #356

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
645688d
IoSerialiserYaS: remove deprecated template keyword
wirew0rm Nov 5, 2024
df7b1ef
RestClient: Correctly use LongPollingIdx Parameter
wirew0rm Nov 5, 2024
ae567da
RestBackend: Remove artificial rate limit
wirew0rm Nov 6, 2024
4e1d52e
majordomoworker_rest_tests: test with LongPollIdx
wirew0rm Nov 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 38 additions & 15 deletions src/client/include/RestClientEmscripten.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,27 @@ using namespace opencmw;
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.
*/
static std::string getFinalURL(std::uint32_t id) {
auto finalURLChar = static_cast<char *>(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));
std::string finalURL{ finalURLChar, strlen(finalURLChar) };
EM_ASM({ _free($0) }, finalURLChar);
return finalURL;
}

struct pointer_equals {
using is_transparent = void;
template<typename Left, typename Right>
Expand Down Expand Up @@ -103,6 +124,7 @@ static std::unordered_set<std::unique_ptr<detail::SubscriptionPayload>, detail::
struct SubscriptionPayload : FetchPayload {
bool _live = true;
MIME::MimeType _mimeType;
std::size_t _update = 0;

SubscriptionPayload(Command &&_command, MIME::MimeType mimeType)
: FetchPayload(std::move(_command))
Expand All @@ -114,7 +136,7 @@ struct SubscriptionPayload : FetchPayload {
SubscriptionPayload &operator=(SubscriptionPayload &&other) noexcept = default;

void requestNext() {
auto uri = command.topic;
auto uri = opencmw::URI<opencmw::STRICT>::UriFactory(command.topic).addQueryParameter("LongPollingIdx", (_update == 0) ? "Next" : fmt::format("{}", _update)).build();
fmt::print("URL 1 >>> {}\n", uri.relativeRef());
auto preferredHeader = detail::getPreferredContentTypeHeader(command.topic, _mimeType);
std::array<const char *, preferredHeader.size() + 1> preferredHeaderEmscripten;
Expand All @@ -140,14 +162,22 @@ struct SubscriptionPayload : FetchPayload {
attr.attributes = EMSCRIPTEN_FETCH_LOAD_TO_MEMORY;
attr.requestHeaders = preferredHeaderEmscripten.data();
attr.onsuccess = [](emscripten_fetch_t *fetch) {
auto payloadIt = getPayloadIt(fetch);
auto &payload = *payloadIt;
auto payloadIt = getPayloadIt(fetch);
auto &payload = *payloadIt;
std::string finalURL = getFinalURL(fetch->id);
std::string longPollingIdxString = opencmw::URI<>(finalURL).queryParamMap().at("LongPollingIdx").value_or("0");
char *end = longPollingIdxString.data() + longPollingIdxString.size();
std::size_t longPollingIdx = strtoull(longPollingIdxString.data(), &end, 10);
if (longPollingIdx != payload->_update) {
fmt::print("received unexpected update: {}, expected {}\n", longPollingIdx, payload->_update);
return;
}
payload->onsuccess(fetch->status, std::string_view(fetch->data, detail::checkedStringViewSize(fetch->numBytes)));
emscripten_fetch_close(fetch);
payload->_update++;
if (payload->_live) {
payload->requestNext();
} else {
emscripten_fetch_close(fetch);
detail::subscriptionPayloads.erase(payloadIt);
}
};
Expand Down Expand Up @@ -195,7 +225,7 @@ class RestClient : public ClientBase {
}
~RestClient() { RestClient::stop(); };

void stop() override{};
void stop() override {};

std::vector<std::string> protocols() noexcept override { return { "http", "https" }; }

Expand Down Expand Up @@ -273,26 +303,19 @@ class RestClient : public ClientBase {
}

void startSubscription(Command &&cmd) {
auto uri = opencmw::URI<>::factory(cmd.topic).addQueryParameter("LongPollingIdx", "Next").build();
cmd.topic = uri;

auto payload = std::make_unique<detail::SubscriptionPayload>(std::move(cmd), _mimeType);
auto rawPayload = payload.get();
detail::subscriptionPayloads.insert(std::move(payload));
fmt::print("starting subscription: {}, existiing subscriptions: {}\n", cmd.topic.str(), detail::subscriptionPayloads.size());
rawPayload->requestNext();
}

void stopSubscription(Command &&cmd) {
// TODO: Can we provide
// Subscription subscribe(...)
// void get(...)
// void set(...)
// instead of going through a fake generic request(...)?
auto uri = opencmw::URI<>::factory(cmd.topic).addQueryParameter("LongPollingIdx", "Next").build();
auto payloadIt = std::find_if(detail::subscriptionPayloads.begin(), detail::subscriptionPayloads.end(),
[&](const auto &ptr) {
return ptr->command.topic == uri;
return ptr->command.topic == cmd.topic;
});
fmt::print("stopping subscription: {}, existiing subscriptions: {}\n", cmd.topic.str(), detail::subscriptionPayloads.size());
if (payloadIt == detail::subscriptionPayloads.end()) {
return;
}
Expand Down
50 changes: 22 additions & 28 deletions src/client/include/RestClientNative.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ class MinIoThreads {
public:
MinIoThreads() = default;
MinIoThreads(int value) noexcept
: _minThreads(value){};
: _minThreads(value) {};
constexpr operator int() const noexcept { return _minThreads; };
};

Expand All @@ -43,7 +43,7 @@ class MaxIoThreads {
public:
MaxIoThreads() = default;
MaxIoThreads(int value) noexcept
: _maxThreads(value){};
: _maxThreads(value) {};
constexpr operator int() const noexcept { return _maxThreads; };
};

Expand All @@ -52,9 +52,9 @@ struct ClientCertificates {

ClientCertificates() = default;
ClientCertificates(const char *X509_ca_bundle) noexcept
: _certificates(X509_ca_bundle){};
: _certificates(X509_ca_bundle) {};
ClientCertificates(const std::string &X509_ca_bundle) noexcept
: _certificates(X509_ca_bundle){};
: _certificates(X509_ca_bundle) {};
constexpr operator std::string() const noexcept { return _certificates; };
};

Expand Down Expand Up @@ -313,37 +313,31 @@ class RestClient : public ClientBase {
{
client.set_follow_location(true);

auto longPollingEndpoint = [&] {
if (!cmd.topic.queryParamMap().contains(LONG_POLLING_IDX_TAG)) {
return URI<>::factory(cmd.topic).addQueryParameter(LONG_POLLING_IDX_TAG, "Next").build();
} else {
return URI<>::factory(cmd.topic).build();
}
}();

const auto pollHeaders = getPreferredContentTypeHeader(longPollingEndpoint);
auto endpoint = longPollingEndpoint.relativeRef().value();
std::size_t longPollingIdx = 0;
const auto pollHeaders = getPreferredContentTypeHeader(cmd.topic);
client.set_read_timeout(cmd.timeout); // default keep-alive value
while (_run) {
auto redirect_get = [&client](auto url, auto headers) {
for (;;) {
auto result = client.Get(url, headers);
if (!result) return result;

if (result->status >= 300 && result->status < 400) {
url = httplib::detail::decode_url(result.value().get_header_value("location"), true);
} else {
return result;
}
Comment on lines -328 to -337
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was not needed, since httplib resolves redirects internally with the set_follow_location(true) setting which is applied above

auto endpoint = [&]() {
if (longPollingIdx == 0UZ) {
return URI<STRICT>::factory(cmd.topic).addQueryParameter(LONG_POLLING_IDX_TAG, "Next").build().relativeRef().value();
} else {
return URI<STRICT>::factory(cmd.topic).addQueryParameter(LONG_POLLING_IDX_TAG, fmt::format("{}", longPollingIdx)).build().relativeRef().value();
}
};
if (const httplib::Result &result = redirect_get(endpoint, pollHeaders)) {
}();
if (const httplib::Result &result = client.Get(endpoint, pollHeaders)) {
returnMdpMessage(cmd, result);
} else { // failed or server is down -> wait until retry
std::this_thread::sleep_for(cmd.timeout); // time-out until potential retry
// update long-polling-index
std::string location = result->location.empty() ? endpoint : result->location;
std::string updateIdxString = URI<STRICT>(location).queryParamMap().at(std::string(LONG_POLLING_IDX_TAG)).value_or("0");
char *end = updateIdxString.data() + updateIdxString.size();
longPollingIdx = strtoull(updateIdxString.data(), &end, 10) + 1;
auto headerTimestamp = result->get_header_value_u64("X-TIMESTAMP");
auto latency = headerTimestamp - std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::high_resolution_clock::now().time_since_epoch()).count();
} else { // failed or server is down -> wait until retry
if (_run) {
returnMdpMessage(cmd, result, fmt::format("Long-Polling-GET request failed for {}: {}", cmd.topic.str(), static_cast<int>(result.error())));
}
std::this_thread::sleep_for(cmd.timeout); // time-out until potential retry
}
}
}
Expand Down
Loading
Loading