Guard last_event_id() against the async event loop in SSEClient (#2644)

last_event_id() returned a reference to a string that run_event_loop() assigns on its own thread under start_async(), so a caller reading it from another thread raced the writes. The member is now guarded by a mutex and the getter returns a copy.

The return type changes from const std::string & to std::string. Code that binds the result to a const reference or copies it still compiles.
This commit is contained in:
metsw24-max authored and GitHub committed 2026-10-25 22:21:55 -04:00
1 parent 391de32115
commit 2f8b9f6f57
3 files changed
+51 -4

No files matched your search

+1 -1
View File
@@ -100,7 +100,7 @@ sse.stop();
```cpp ```cpp
bool connected = sse.is_connected(); bool connected = sse.is_connected();
const std::string &id = sse.last_event_id(); std::string id = sse.last_event_id();
``` ```
## Examples ## Examples
+9 -3
View File
@@ -4498,7 +4498,7 @@ public:
// State accessors // State accessors
bool is_connected() const; bool is_connected() const;
const std::string &last_event_id() const; std::string last_event_id() const;
// Blocking start - runs event loop with auto-reconnect // Blocking start - runs event loop with auto-reconnect
void start(); void start();
@@ -4536,7 +4536,9 @@ private:
// State // State
std::atomic<bool> running_{false}; std::atomic<bool> running_{false};
std::atomic<bool> connected_{false}; std::atomic<bool> connected_{false};
// Written by the event loop, which start_async() runs on its own thread
std::string last_event_id_; std::string last_event_id_;
mutable std::mutex last_event_id_mutex_;
// Async support // Async support
std::thread async_thread_; std::thread async_thread_;
@@ -5034,7 +5036,8 @@ inline SSEClient &SSEClient::set_headers(const Headers &headers) {
inline bool SSEClient::is_connected() const { return connected_.load(); } inline bool SSEClient::is_connected() const { return connected_.load(); }
inline const std::string &SSEClient::last_event_id() const { inline std::string SSEClient::last_event_id() const {
std::lock_guard<std::mutex> lock(last_event_id_mutex_);
return last_event_id_; return last_event_id_;
} }
@@ -5209,7 +5212,10 @@ inline void SSEClient::run_event_loop() {
if (event_complete) { if (event_complete) {
// Update last_event_id for reconnection, even for an event that // Update last_event_id for reconnection, even for an event that
// has no data. An empty id clears it. // has no data. An empty id clears it.
if (has_id) { last_event_id_ = current_msg.id; } if (has_id) {
std::lock_guard<std::mutex> lock(last_event_id_mutex_);
last_event_id_ = current_msg.id;
}
// An event without a data field is not dispatched // An event without a data field is not dispatched
if (has_data) { dispatch_event(current_msg); } if (has_data) { dispatch_event(current_msg); }
+41
View File
@@ -26459,6 +26459,47 @@ TEST_F(SSEIntegrationTest, LastEventIdHeader) {
EXPECT_EQ(sse.last_event_id(), "evt-123"); EXPECT_EQ(sse.last_event_id(), "evt-123");
} }
// Test: last_event_id() read from another thread while start_async() runs
TEST_F(SSEIntegrationTest, LastEventIdWhileRunning) {
// Ids of every length up to 200, so the stored id keeps changing size
server_->Get("/many-ids", [](const Request &, Response &res) {
res.set_chunked_content_provider(
"text/event-stream", [](size_t offset, DataSink &sink) {
if (offset == 0) {
std::string events;
for (size_t i = 0; i < 2000; i++) {
events += "id: " + std::string(1 + i % 200, 'a') + "\n";
events += "data: x\n\n";
}
events += "id: end\ndata: x\n\n";
sink.write(events.data(), events.size());
}
return false;
});
});
Client client("localhost", get_port());
sse::SSEClient sse(client, "/many-ids");
// Wait out the rest of the test rather than replay the stream
sse.set_reconnect_interval(10000);
sse.set_max_reconnect_attempts(1);
sse.start_async();
auto well_formed = true;
std::string id;
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
while (id != "end" && std::chrono::steady_clock::now() < deadline) {
id = sse.last_event_id();
if (id != "end" && id.find_first_not_of('a') != std::string::npos) {
well_formed = false;
}
}
sse.stop();
EXPECT_TRUE(well_formed);
EXPECT_EQ(sse.last_event_id(), "end");
}
// Test: Manual stop // Test: Manual stop
TEST_F(SSEIntegrationTest, ManualStop) { TEST_F(SSEIntegrationTest, ManualStop) {
// Endpoint that sends one event and stays open briefly. The server outlives // Endpoint that sends one event and stays open briefly. The server outlives