tests: processing incomplete events in SSE client

During my testing I noticed that I obtained incomplete message in
on_data callback. The message was split into two callbacks.
This commit fixes the FIXME by buffering the data and processing
a complete event, once available.

Change-Id: Ied07e69e8b518f20fcc82134a4c041e7ec3a06d6
Signed-off-by: Mattias Walström <lazzer@gmail.com>

Author: peckato1

tests/restconf_utils.cpp

@@ -199,8 +199,8 @@ SSEClient::SSEClient(
         req->on_response([&, silenceTimeout, reportIgnoredLines](const ng_client::response& res) {
             requestSent.release();
             res.on_data([&, silenceTimeout, reportIgnoredLines](const uint8_t* data, std::size_t len) {
-                // not a production-ready code. In real-life condition the data received in one callback might probably be incomplete
-                parseEvents(std::string(reinterpret_cast<const char*>(data), len), eventWatcher, reportIgnoredLines);
+                dataBuffer.append(std::string(reinterpret_cast<const char*>(data), len));
+                parseEvents(eventWatcher, reportIgnoredLines);
                 t.expires_from_now(silenceTimeout);
             });
         });
@@ -217,30 +217,39 @@ SSEClient::SSEClient(
     });
 }
 
-void SSEClient::parseEvents(const std::string& msg, const RestconfNotificationWatcher& eventWatcher, const ReportIgnoredLines reportIgnoredLines)
+void SSEClient::parseEvents(const RestconfNotificationWatcher& eventWatcher, const ReportIgnoredLines reportIgnoredLines)
 {
     static const std::string dataPrefix = "data:";
     static const std::string ignorePrefix = ":";
 
-    std::istringstream iss(msg);
-    std::string line;
-    std::string event;
-
-    while (std::getline(iss, line)) {
-        if (line.starts_with(ignorePrefix) && reportIgnoredLines == ReportIgnoredLines::Yes) {
-            eventWatcher.commentEvent(line);
-        } else if (line.starts_with(ignorePrefix)) {
-            continue;
-        } else if (line.starts_with(dataPrefix)) {
-            event += line.substr(dataPrefix.size());
-        } else if (line.empty() && !event.empty()) {
-            eventWatcher.dataEvent(event);
-            event.clear();
-        } else if (line.empty()) {
-            continue;
-        } else {
-            CAPTURE(msg);
-            FAIL("Unprefixed response");
+    std::size_t pos = 0;
+    constexpr auto EVENT_SEPARATOR = "\n\n"; // FIXME: Not a production-ready code; does not deal with all possible newline combinations of CR and LF
+
+    while ((pos = dataBuffer.find(EVENT_SEPARATOR)) != std::string::npos) {
+        // extract event
+        auto rawEvent = dataBuffer.substr(0, pos + std::char_traits<char>::length(EVENT_SEPARATOR));
+        std::istringstream stream(rawEvent);
+        dataBuffer.erase(0, pos + std::char_traits<char>::length(EVENT_SEPARATOR));
+
+        // split on newlines
+        std::string line;
+        std::string event;
+        while (std::getline(stream, line)) {
+            if (line.starts_with(ignorePrefix) && reportIgnoredLines == ReportIgnoredLines::Yes) {
+                eventWatcher.commentEvent(line);
+            } else if (line.starts_with(ignorePrefix)) {
+                continue;
+            } else if (line.starts_with(dataPrefix)) {
+                event += line.substr(dataPrefix.size());
+            } else if (line.empty() && !event.empty()) {
+                eventWatcher.dataEvent(event);
+                event.clear();
+            } else if (line.empty()) {
+                continue;
+            } else {
+                CAPTURE(rawEvent);
+                FAIL("Unprefixed response");
+            }
         }
     }
 }

tests/restconf_utils.h

@@ -87,6 +87,7 @@ void setupRealNacm(sysrepo::Session session);
 struct SSEClient {
     std::shared_ptr<ng_client::session> client;
     boost::asio::deadline_timer t;
+    std::string dataBuffer;
 
     enum class ReportIgnoredLines {
         No,
@@ -104,7 +105,7 @@ struct SSEClient {
         const boost::posix_time::seconds silenceTimeout = boost::posix_time::seconds(1), // test code; the server should respond "soon"
         const ReportIgnoredLines reportIgnoredLines = ReportIgnoredLines::No);
 
-    static void parseEvents(const std::string& msg, const RestconfNotificationWatcher& eventWatcher, const ReportIgnoredLines reportIgnoredLines);
+    void parseEvents(const RestconfNotificationWatcher& eventWatcher, const ReportIgnoredLines reportIgnoredLines);
 };
 
 #define PREPARE_LOOP_WITH_EXCEPTIONS \