Consolidate parsing.

Author: kinyoklion

libs/server-sent-events/src/curl_client.cpp

@@ -232,10 +232,9 @@ bool CurlClient::SetupCurlOptions(CURL* curl,
         headers = curl_slist_append(headers, header.c_str());
     }
 
-    // Add Last-Event-ID if we have one
+    // Add Last-Event-ID if we have one from previous connection
     if (context.last_event_id && !context.last_event_id->empty()) {
-        std::string last_event_header =
-            "Last-Event-ID: " + *context.last_event_id;
+        std::string last_event_header = "Last-Event-ID: " + *context.last_event_id;
         headers = curl_slist_append(headers, last_event_header.c_str());
     }
 
@@ -322,7 +321,7 @@ int CurlClient::ProgressCallback(void* clientp,
             context->last_progress_time = now;
         } else {
             // No new data - check if we've exceeded the timeout
-            auto elapsed = std::chrono::duration_cast<
+            const auto elapsed = std::chrono::duration_cast<
                 std::chrono::milliseconds>(
                 now - context->last_progress_time);
 
@@ -369,128 +368,17 @@ size_t CurlClient::WriteCallback(const char* data,
         return 0; // Abort the transfer
     }
 
-    // Parse SSE data
-    std::string_view body(data, total_size);
-
-    // Parse stream into lines
-    size_t i = 0;
-    while (i < body.size()) {
-        // Find next line delimiter
-        const size_t delimiter_pos = body.find_first_of("\r\n", i);
-        const size_t append_size = (delimiter_pos == std::string::npos)
-                                       ? (body.size() - i)
-                                       : (delimiter_pos - i);
-
-        // Append to buffered line
-        if (context->buffered_line.has_value()) {
-            context->buffered_line->append(body.substr(i, append_size));
-        } else {
-            context->buffered_line = std::string(body.substr(i, append_size));
-        }
-
-        i += append_size;
-
-        if (i >= body.size()) {
-            break;
-        }
-
-        // Handle line delimiters
-        if (body[i] == '\r') {
-            context->complete_lines.push_back(*context->buffered_line);
-            context->buffered_line.reset();
-            context->begin_CR = true;
-            i++;
-        } else if (body[i] == '\n') {
-            if (context->begin_CR) {
-                context->begin_CR = false;
-            } else {
-                context->complete_lines.push_back(*context->buffered_line);
-                context->buffered_line.reset();
-            }
-            i++;
-        }
-    }
-
-    // Parse completed lines into events
-    while (!context->complete_lines.empty()) {
-        std::string line = std::move(context->complete_lines.front());
-        context->complete_lines.pop_front();
-
-        if (line.empty()) {
-            // Empty line indicates end of event
-            if (context->current_event) {
-                // Trim trailing newline from data
-                if (!context->current_event->data.empty() &&
-                    context->current_event->data.back() == '\n') {
-                    context->current_event->data.pop_back();
-                }
-
-                // Update last_event_id_ only when dispatching a completed event
-                if (context->current_event->id) {
-                    context->last_event_id = context->current_event->id;
-                }
-
-                // Dispatch event on executor thread
-                auto event_data = context->current_event->data;
-                auto event_type = context->current_event->type.empty()
-                                      ? "message"
-                                      : context->current_event->type;
-                auto event_id = context->current_event->id;
-                context->receive(Event(
-                    std::move(event_type),
-                    std::move(event_data),
-                    std::move(event_id)));
-
-                context->current_event.reset();
-            }
-            continue;
-        }
-
-        // Parse field
-        const size_t colon_pos = line.find(':');
-        if (colon_pos == 0) {
-            // Comment line, dispatch it
-            std::string comment = line.substr(1);
-
-            context->receive(Event("comment", comment));
-            continue;
-        }
-
-        std::string field_name;
-        std::string field_value;
-
-        if (colon_pos == std::string::npos) {
-            field_name = line;
-            field_value = "";
-        } else {
-            field_name = line.substr(0, colon_pos);
-            field_value = line.substr(colon_pos + 1);
-
-            // Remove leading space from value if present
-            if (!field_value.empty() && field_value[0] == ' ') {
-                field_value = field_value.substr(1);
-            }
-        }
-
-        // Initialize event if needed
-        if (!context->current_event) {
-            context->current_event.emplace(detail::Event{});
-            context->current_event->id = context->last_event_id;
+    // Set up the event receiver callback for the parser
+    context->parser_body->on_event([context](Event event) {
+        // Track last event ID for reconnection
+        if (event.id()) {
+            context->last_event_id = event.id();
         }
+        context->receive(std::move(event));
+    });
 
-        // Handle field
-        if (field_name == "event") {
-            context->current_event->type = field_value;
-        } else if (field_name == "data") {
-            context->current_event->data += field_value;
-            context->current_event->data += '\n';
-        } else if (field_name == "id") {
-            if (field_value.find('\0') == std::string::npos) {
-                context->current_event->id = field_value;
-            }
-        }
-        // retry field is ignored for now
-    }
+    const std::string_view data_view(data, total_size);
+    context->parser_reader->put(data_view);
 
     return total_size;
 }
@@ -525,11 +413,8 @@ void CurlClient::PerformRequestWithMulti(
         return;
     }
 
-    // Clear parser state for new connection
-    context->buffered_line.reset();
-    context->complete_lines.clear();
-    context->current_event.reset();
-    context->begin_CR = false;
+    // Initialize parser for new connection (last_event_id is tracked separately)
+    context->init_parser();
 
     CURL* curl = curl_easy_init();
     if (!curl) {

libs/server-sent-events/src/curl_client.hpp

@@ -73,19 +73,18 @@ class CurlClient final : public Client,
         std::optional<Callbacks> callbacks_;
 
     public:
-        // SSE parser state
-        std::optional<std::string> buffered_line;
-        std::deque<std::string> complete_lines;
-        bool begin_CR;
+        // SSE parser using common parser from parser.hpp
+        using ParserBody = detail::EventBody<std::function<void(Event)>>;
+        std::unique_ptr<typename ParserBody::value_type> parser_body;
+        std::unique_ptr<typename ParserBody::reader> parser_reader;
+
+        // Track last event ID for reconnection (separate from parser state)
+        std::optional<std::string> last_event_id;
 
         // Progress tracking for read timeout
         std::chrono::steady_clock::time_point last_progress_time;
         curl_off_t last_download_amount;
 
-
-        std::optional<std::string> last_event_id;
-        std::optional<detail::Event> current_event;
-
         const http::request<http::string_body> req;
         const std::string url;
         const std::optional<std::chrono::milliseconds> connect_timeout;
@@ -182,11 +181,7 @@ class CurlClient final : public Client,
                        bool skip_verify_peer
             ) : shutting_down_(false),
                 curl_socket_(CURL_SOCKET_BAD),
-                buffered_line(std::nullopt),
-                begin_CR(false),
                 last_download_amount(0),
-                last_event_id(std::nullopt),
-                current_event(std::nullopt),
                 req(std::move(req)),
                 url(std::move(url)),
                 connect_timeout(connect_timeout),
@@ -196,6 +191,12 @@ class CurlClient final : public Client,
                 proxy_url(std::move(proxy_url)),
                 skip_verify_peer(skip_verify_peer) {
         }
+
+        void init_parser() {
+            parser_body = std::make_unique<typename ParserBody::value_type>();
+            parser_reader = std::make_unique<typename ParserBody::reader>(*parser_body);
+            parser_reader->init();
+        }
     };
 
 public:

libs/server-sent-events/src/parser.hpp

@@ -62,6 +62,16 @@ struct EventBody<EventReceiver>::reader {
     std::optional<Event> event_;
 
    public:
+    // Constructor for standalone use (curl_client) - no Boost types required
+    explicit reader(value_type& body)
+        : body_(body),
+          buffered_line_(),
+          complete_lines_(),
+          begin_CR_(false),
+          event_() {
+    }
+
+    // Constructor for Boost Beast HTTP body reader (FoxyClient)
     template <bool isRequest, class Fields>
     reader(http::header<isRequest, Fields>& h, value_type& body)
         : body_(body),
@@ -87,6 +97,11 @@ struct EventBody<EventReceiver>::reader {
         ec = {};
     }
 
+    // Simplified init for standalone use - no Boost types required
+    void init() {
+        // Nothing to initialize
+    }
+
     /**
      * Store buffers.
      * This is called zero or more times with parsed body octets.
@@ -104,6 +119,16 @@ struct EventBody<EventReceiver>::reader {
         return buffer_bytes(buffers);
     }
 
+    /**
+     * Simplified put for standalone use - no Boost types required.
+     * Feed data into the parser. This can be called multiple times as data arrives.
+     * @param data The data to parse
+     */
+    void put(std::string_view data) {
+        parse_stream(data);
+        parse_events();
+    }
+
     /**
      * Called when the body is complete.
      * @param ec Set to the error, if any occurred.
@@ -124,20 +149,20 @@ struct EventBody<EventReceiver>::reader {
     // Appends the body to the buffered line until reaching any of the
     // characters specified within the search parameter. The search parameter is
     // treated as an array of search characters, not as a single token.
-    size_t append_up_to(boost::string_view body, std::string const& search) {
+    size_t append_up_to(std::string_view body, std::string const& search) {
         std::size_t index = body.find_first_of(search);
         if (index != std::string::npos) {
             body.remove_suffix(body.size() - index);
         }
         if (buffered_line_.has_value()) {
-            buffered_line_->append(body.to_string());
+            buffered_line_->append(body);
         } else {
             buffered_line_ = std::string{body};
         }
         return index == std::string::npos ? body.size() : index;
     }
 
-    void parse_stream(boost::string_view body) {
+    void parse_stream(std::string_view body) {
         size_t i = 0;
         while (i < body.size()) {
             i += this->append_up_to(body.substr(i, body.length() - i), "\r\n");