diff --git a/README-sse.md b/README-sse.md index 8f799755..e402c2cf 100644 --- a/README-sse.md +++ b/README-sse.md @@ -100,7 +100,7 @@ sse.stop(); ```cpp bool connected = sse.is_connected(); -const std::string &id = sse.last_event_id(); +std::string id = sse.last_event_id(); ``` ## Examples diff --git a/httplib.h b/httplib.h index 76960c80..5ed34824 100644 --- a/httplib.h +++ b/httplib.h @@ -4498,7 +4498,7 @@ public: // State accessors 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 void start(); @@ -4536,7 +4536,9 @@ private: // State std::atomic running_{false}; std::atomic connected_{false}; + // Written by the event loop, which start_async() runs on its own thread std::string last_event_id_; + mutable std::mutex last_event_id_mutex_; // Async support 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 const std::string &SSEClient::last_event_id() const { +inline std::string SSEClient::last_event_id() const { + std::lock_guard lock(last_event_id_mutex_); return last_event_id_; } @@ -5209,7 +5212,10 @@ inline void SSEClient::run_event_loop() { if (event_complete) { // Update last_event_id for reconnection, even for an event that // has no data. An empty id clears it. - if (has_id) { last_event_id_ = current_msg.id; } + if (has_id) { + std::lock_guard lock(last_event_id_mutex_); + last_event_id_ = current_msg.id; + } // An event without a data field is not dispatched if (has_data) { dispatch_event(current_msg); } diff --git a/test/test.cc b/test/test.cc index fa0bbb2c..0087e5fe 100644 --- a/test/test.cc +++ b/test/test.cc @@ -26459,6 +26459,47 @@ TEST_F(SSEIntegrationTest, LastEventIdHeader) { 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_F(SSEIntegrationTest, ManualStop) { // Endpoint that sends one event and stays open briefly. The server outlives