Rest client catch up

cmake/Sodium.cmake

@@ -11,7 +11,9 @@ file(GLOB_RECURSE SODIUM_C "${libsodium_SOURCE_DIR}/src/*.c")
 file(GLOB_RECURSE SODIUM_H "${libsodium_SOURCE_DIR}/src/*.h")
 
 add_library(sodium STATIC ${SODIUM_C} ${SODIUM_H})
-
+if(CMAKE_C_COMPILER_ID MATCHES "GNU|Clang")
+    target_compile_options(sodium PRIVATE -w)
+endif()
 target_include_directories(sodium PUBLIC $<BUILD_INTERFACE:${libsodium_SOURCE_DIR}/src/libsodium/include> $<INSTALL_INTERFACE:include/opencmw> PRIVATE "${libsodium_SOURCE_DIR}/src/libsodium/include/sodium")
 # silence warnings about being built by a different build system
 target_compile_definitions(sodium PRIVATE CONFIGURED)

docs/RestUriMapping.md

@@ -18,10 +18,69 @@ The REST backend maps HTTP requests to Majordomo requests.
 By default, `PUT` and `POST` HTTP requests are mapped to Majordomo's `Set`,
 and `GET` HTTP request is mapped to Majordomo's `Get` request.
 
-Requests with a query parameter `LongPollingIdx` are treated as `LongPoll` request subscribing to the
-given topic (consisting of URI path and other query parameters), with the possible values:
+Use `LongPollingIdx` on a GET request to read notifications for a topic
+(the path and other query parameters):
 
- - `Next`: Redirects to the next notification message that arrives after the request has been processed.
- - `Last`: Redirects to the most recent notification message that is in the cache when the request is
- processed. If there is no such entry yet, it's treated like `Next`, i.e. waits for the next notification.
- - a positive integer value, to retrieve a specific cache entry.
+- `Next`: Redirect to the next new message's index.
+- `Last`: Redirect to the newest buffered message, or use `Next` if the buffer is empty.
+- A non-negative integer: Read that message, waiting if it has not arrived.
+
+Add `LongPollingBatch` to receive several messages in one HTTP response:
+
+- `?LongPollingIdx=42&LongPollingBatch=10`:
+  Wait for messages 42 through 51, then return all ten.
+- `?LongPollingIdx=42&LongPollingBatch=AllAvailable`:
+  Return message 42 and all newer buffered messages. If 42 has not arrived, wait for it.
+- `?LongPollingIdx=42` (no batch):
+  Return only message 42, waiting if needed.
+
+The server keeps the latest 100 messages per subscription. If the requested start is too old, a batch starts
+at the oldest buffered message. A numeric batch still waits for exactly the requested count.
+Without a batch, a too-old index redirects to the next message, as before.
+
+Clients report detected gaps in `Message::error`, alongside the next valid sample.
+The warning does not invalidate that sample or stop the subscription.
+
+`LongPollingBatch` requires `LongPollingIdx`. The batch size must be 1–100 or `AllAvailable`;
+it does not mean "the latest N messages" or a time range. Invalid values and index ranges too
+large to represent return HTTP `400`.
+
+Batches are intended for direct connections. Proxy caching is unchanged.
+
+If a successful batch response cannot be decoded, the client reports an error and continues after its
+last index, if known; otherwise it uses `Next`, which may skip buffered messages.
+HTTP `504` retries the same index if known. Other HTTP errors stop the batch subscription.
+
+Batch responses use `multipart/mixed`, with one part per message. Each part carries its own
+index, topic, service name, and payload length. The outer `x-opencmw-long-polling-idx` header
+gives the first returned index; topic and service name appear only in the parts.
+
+Example with two messages. All line breaks are `\r\n`, including after the closing boundary:
+
+```text
+--opencmw-long-polling-multipart-boundary
+x-opencmw-long-polling-idx: 42
+x-opencmw-topic: /colors?sample=42
+x-opencmw-service-name: colors-service
+content-length: 5
+
+hello
+--opencmw-long-polling-multipart-boundary
+x-opencmw-long-polling-idx: 43
+x-opencmw-topic: /colors?sample=43
+x-opencmw-service-name: colors-service
+content-length: 5
+
+world
+--opencmw-long-polling-multipart-boundary--
+```
+
+Response header:
+`content-type: multipart/mixed; boundary=opencmw-long-polling-multipart-boundary`.
+
+The boundary is fixed. Read each payload by its `content-length`, not by searching for the
+boundary. The length counts payload bytes only, excluding the following line break.
+
+Generic MIME parsers can also be used, but they find parts by searching for boundary delimiter
+lines rather than using `content-length`. We do not guarantee that the boundary is absent from
+payloads. A matching delimiter line inside a payload may therefore be mistaken for the end of a part.

src/client/include/RestClientEmscripten.hpp

@@ -14,6 +14,7 @@
 #include <concepts>
 #include <cstdint>
 #include <cstdio>
+#include <expected>
 #include <format>
 #include <functional>
 #include <iostream>
@@ -34,7 +35,9 @@
 
 #include <ClientCommon.hpp>
 #include <ClientContext.hpp>
+#include <LongPollingBatch.hpp>
 #include <MIME.hpp>
+#include <Topic.hpp>
 #include <URI.hpp>
 
 using namespace opencmw;
@@ -72,6 +75,49 @@
     }
 }
 
+inline bool usesLongPollingBatch(const mdp::Message::URI &topic) {
+    return topic.queryParamMap().contains(std::string(long_polling::kBatchParameter));
+}
+
+inline std::expected<std::string, std::string> takeInitialLongPollingIndex(Command &command) {
+    auto       parameters = command.topic.queryParamMap();
+    const auto entry      = parameters.find(std::string(long_polling::kIndexParameter));
+    if (entry == parameters.end()) {
+        return "Next";
+    }
+    if (!entry->second.has_value()) {
+        return std::unexpected("LongPollingIdx requires a value");
+    }
+
+    std::string index = *entry->second;
+    if (index != "Next" && index != "Last") {
+        auto parsed = long_polling::parseUnsigned(*entry->second, long_polling::kIndexParameter);
+        if (!parsed.has_value()) {
+            return std::unexpected(parsed.error());
+        }
+        index = std::to_string(*parsed);
+    }
+
+    parameters.erase(entry);
+    command.topic = mdp::Message::URI::UriFactory(command.topic).setQuery(std::move(parameters)).build();
+    return index;
+}
+
+inline mdp::Topic subscriptionTopic(const mdp::Message::URI &topic) {
+    auto parameters = topic.queryParamMap();
+    parameters.erase(std::string(long_polling::kIndexParameter));
+    parameters.erase(std::string(long_polling::kBatchParameter));
+    const auto normalized = mdp::Message::URI::UriFactory(topic).setQuery(std::move(parameters)).build();
+    return mdp::Topic::fromMdpTopic(normalized);
+}
+
+inline bool sameSubscription(const mdp::Message::URI &lhs, const mdp::Message::URI &rhs) {
+    return lhs.scheme() == rhs.scheme()
+        && lhs.hostName() == rhs.hostName()
+        && lhs.port() == rhs.port()
+        && subscriptionTopic(lhs) == subscriptionTopic(rhs);
+}
+
 struct SubscriptionState {
     Command                      command{};
     std::optional<std::uint64_t> lastDeliveredIndex{};
@@ -129,14 +175,30 @@
     }
 
     void startSubscription(Command &&cmd) {
-        const std::uint64_t id = _nextSubscriptionId++;
-        _subscriptions.emplace(id, SubscriptionState{ .command = std::move(cmd) });
-        startNextLongPoll(id, std::nullopt);
+        const auto initialIndex = takeInitialLongPollingIndex(cmd);
+        if (!initialIndex.has_value()) {
+            reportFailure(cmd, initialIndex.error());
+            return;
+        }
+        const bool usesBatch = usesLongPollingBatch(cmd.topic);
+        if (std::ranges::any_of(_subscriptions, [&](const auto &entry) {
+                return sameSubscription(entry.second.command.topic, cmd.topic)
+                    && (usesBatch || usesLongPollingBatch(entry.second.command.topic));
+            })) {
+            reportFailure(cmd, "A subscription for this topic is already active; concurrent batch subscriptions are not supported");
+            return;
+        }
+        const std::uint64_t id           = _nextSubscriptionId++;
+        auto              &subscription = _subscriptions.emplace(id, SubscriptionState{ .command = std::move(cmd) }).first->second;
+        if (const auto index = long_polling::parseUnsigned(*initialIndex, long_polling::kIndexParameter); index.has_value() && *index > 0) {
+            subscription.lastDeliveredIndex = *index - 1; // Detect a gap before the first returned sample.
+        }
+        startLongPoll(id, *initialIndex);
     }
 
     void stopSubscription(const Command &cmd) {
         const auto entry = std::ranges::find_if(_subscriptions,
-                [&](const auto &pair) { return pair.second.command.topic == cmd.topic; });
+                [&](const auto &pair) { return sameSubscription(pair.second.command.topic, cmd.topic); });
         if (entry == _subscriptions.end()) {
             return;
         }
@@ -147,7 +209,7 @@
         }
     }
 
-    void startNextLongPoll(std::uint64_t subscriptionId, std::optional<std::uint64_t> index) noexcept {
+    void startLongPoll(std::uint64_t subscriptionId, std::string longPollingIndex) noexcept {
         try {
             if (!_acceptWork.load(std::memory_order_acquire)) {
                 return;
@@ -156,13 +218,12 @@
             if (entry == _subscriptions.end()) {
                 return;
             }
-            const std::string longPollingIndex = index.has_value() ? std::to_string(*index) : "Next";
 
-            auto              activeFetch      = std::make_unique<ActiveFetch>();
-            activeFetch->owner                 = this;
-            activeFetch->id                    = _nextFetchId++;
-            activeFetch->subscriptionId        = subscriptionId;
-            entry->second.activeFetchId        = activeFetch->id;
+            auto activeFetch            = std::make_unique<ActiveFetch>();
+            activeFetch->owner          = this;
+            activeFetch->id             = _nextFetchId++;
+            activeFetch->subscriptionId = subscriptionId;
+            entry->second.activeFetchId = activeFetch->id;
             startFetch(std::move(activeFetch), URI<STRICT>::UriFactory(entry->second.command.topic).addQueryParameter("LongPollingIdx", longPollingIndex).build());
         } catch (const std::exception &e) {
             endSubscription(subscriptionId, nullptr, 500, {}, e.what());
@@ -171,6 +232,10 @@
         }
     }
 
+    void startNextLongPoll(std::uint64_t subscriptionId, std::optional<std::uint64_t> index) noexcept {
+        startLongPoll(subscriptionId, index.has_value() ? std::to_string(*index) : "Next");
+    }
+
     void startGetOrSet(Command &&cmd) {
         const URI<STRICT> uri         = cmd.topic;
 
@@ -270,63 +335,115 @@
         }
     }
 
     void handleSubscriptionCompletion(std::uint64_t fetchId, std::uint64_t subscriptionId, emscripten_fetch_t *fetch, std::optional<std::string_view> 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;
 
         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{});
 
         // 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;
         }
 
         if (fetchError.has_value()) {
             endSubscription(subscriptionId, fetch, status, body, *fetchError);
             return;
         }
 
+        if (usesLongPollingBatch(state.command.topic)) {
+            const auto parts = long_polling::decodeBatch(body);
+            if (!parts.has_value()) {
+                resumeSubscriptionAfterBatchError(fetchId, subscriptionId, fetch, std::format("could not decode long-polling batch: {}", parts.error()));
+                return;
+            }
+
+            auto                      lastDelivered = state.lastDeliveredIndex;
+            std::vector<mdp::Message> messages;
+            messages.reserve(parts->size());
+            try {
+                for (const auto &part : *parts) {
+                    if (lastDelivered.has_value() && part.index <= *lastDelivered) {
+                        continue;
+                    }
+                    std::string skippedWarning;
+                    if (lastDelivered.has_value() && part.index - *lastDelivered > 1) {
+                        skippedWarning = std::format("Warning: skipped {} samples", part.index - *lastDelivered - 1);
+                    }
+                    auto message        = buildMessage(state.command, status, part.payload, skippedWarning);
+                    message.topic       = mdp::Message::URI(std::string(part.topic));
+                    message.serviceName = std::string(part.serviceName);
+                    messages.push_back(std::move(message));
+                    lastDelivered = part.index;
+                }
+            } catch (const std::exception &e) {
+                const auto lastIndex = parts->back().index;
+                resumeSubscriptionAfterBatchError(fetchId, subscriptionId, fetch, std::format("could not parse long-polling batch metadata: {}", e.what()),
+                        lastIndex < std::numeric_limits<std::uint64_t>::max() ? std::optional{ lastIndex + 1 } : std::nullopt);
+                return;
+            }
+
+            state.lastDeliveredIndex = lastDelivered;
+            const auto callback      = state.command.callback;
+            closeFetch(fetchId, fetch);
+            for (const auto &message : messages) {
+                invokeGuarded(callback, message);
+            }
+            if (lastDelivered.has_value()) {
+                startNextLongPoll(subscriptionId, *lastDelivered + 1);
+            }
+            return;
+        }
+
         if (!index.has_value()) {
             endSubscription(subscriptionId, fetch, status, body, "missing or unparsable LongPollingIdx in the response URL");
             return;
         }
 
         if (state.lastDeliveredIndex.has_value() && *index <= *state.lastDeliveredIndex) {
             const std::uint64_t expected = *state.lastDeliveredIndex + 1;
             closeFetch(fetchId, fetch);
             startNextLongPoll(subscriptionId, expected);
             return;
         }
 
         std::string skippedWarning;
         if (state.lastDeliveredIndex.has_value() && *index - *state.lastDeliveredIndex > 1) {
             skippedWarning = std::format("Warning: skipped {} samples", *index - *state.lastDeliveredIndex - 1);
         }
 
         const mdp::Message message = buildMessage(state.command, status, body, skippedWarning);
         state.lastDeliveredIndex   = *index;
 
         closeFetch(fetchId, fetch);
         invokeGuarded(state.command.callback, message);
         startNextLongPoll(subscriptionId, *index + 1);
     }
 
+    void resumeSubscriptionAfterBatchError(std::uint64_t fetchId, std::uint64_t subscriptionId, emscripten_fetch_t *fetch, std::string_view error, std::optional<std::uint64_t> nextIndex = std::nullopt) {
+        const auto command = _subscriptions.at(subscriptionId).command;
+        const auto status  = fetch->status;
+        closeFetch(fetchId, fetch);
+        reportFailure(command, error, status);
+        startNextLongPoll(subscriptionId, nextIndex);
+    }
+
     void handleGetOrSetCompletion(std::uint64_t fetchId, emscripten_fetch_t *fetch, std::optional<std::string_view> fetchError) {
         if (!_acceptWork.load(std::memory_order_acquire)) {
             closeFetch(fetchId, fetch);
@@ -408,13 +525,13 @@
         emscripten_runtime_keepalive_pop();
     }
 
-    void        reportFailure(const Command &command, std::string_view error) noexcept {
+    void reportFailure(const Command &command, std::string_view error, unsigned short status = 500) noexcept {
         if (!command.callback) {
             std::println(std::cerr, "RestClientEmscripten: {}", error);
             return;
         }
         try {
-            invokeGuarded(command.callback, buildMessage(command, 500, {}, error));
+            invokeGuarded(command.callback, buildMessage(command, status, {}, error));
         } catch (const std::exception &e) {
             std::println(std::cerr, "RestClientEmscripten: could not report '{}': {}", error, e.what());
         } catch (...) {