diff --git a/httplib.h b/httplib.h index 8881eda1..25eade33 100644 --- a/httplib.h +++ b/httplib.h @@ -4493,6 +4493,10 @@ private: std::atomic connected_{false}; std::string last_event_id_; + // Socket of the open stream, for stop() to shut down + std::mutex stream_mutex_; + socket_t stream_sock_ = INVALID_SOCKET; + // Async support std::thread async_thread_; }; @@ -5000,6 +5004,20 @@ inline void SSEClient::start_async() { inline void SSEClient::stop() { running_.store(false); + { + // Ends a blocked read. The loop withdraws the socket under this lock + // before closing it, so it is still open here. + std::lock_guard lock(stream_mutex_); + if (stream_sock_ != INVALID_SOCKET) { +#ifdef _WIN32 + ::shutdown(stream_sock_, SD_BOTH); +#else + // Only the read side, so that the alert TLS sends on the unexpected EOF + // does not raise SIGPIPE + ::shutdown(stream_sock_, SHUT_RD); +#endif + } + } client_.stop(); // Cancel any pending operations if (async_thread_.joinable()) { async_thread_.join(); } } @@ -5073,6 +5091,7 @@ inline void SSEClient::run_event_loop() { // bounded instead of the whole body. auto handle = client_.open_stream("GET", path_, {}, request_headers); handle.body_reader_.payload_max_length = 0; + auto sock = handle.connection_ ? handle.connection_->sock : INVALID_SOCKET; stream::Result result{std::move(handle)}; // Connection error handling @@ -5105,6 +5124,19 @@ inline void SSEClient::run_event_loop() { continue; } + // Hand the socket to stop(), unless it has already been called + { + std::lock_guard lock(stream_mutex_); + if (!running_.load()) { break; } + stream_sock_ = sock; + } + + // Destroyed before result, so the socket is withdrawn before it is closed + detail::scope_exit withdraw_sock([this] { + std::lock_guard lock(stream_mutex_); + stream_sock_ = INVALID_SOCKET; + }); + // Connection successful connected_.store(true); reconnect_count = 0; diff --git a/test/test.cc b/test/test.cc index ff1bf00d..5884f512 100644 --- a/test/test.cc +++ b/test/test.cc @@ -25469,6 +25469,78 @@ TEST_F(SSEIntegrationTest, ManualStop) { EXPECT_FALSE(sse.is_connected()); } +// Sends one event, then nothing until stop is set +static Server::Handler idle_event_stream(std::atomic &stop) { + return [&stop](const Request &, Response &res) { + res.set_chunked_content_provider( + "text/event-stream", [&stop](size_t offset, DataSink &sink) { + if (offset == 0) { + std::string event = "data: hello\n\n"; + sink.write(event.data(), event.size()); + } + while (!stop.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return false; + }); + }; +} + +// Both stop() and the destructor end a read blocked on /idle-stream promptly +static void check_stop_interrupts_blocked_read(Client &client) { + client.set_read_timeout(30, 0); + + for (auto by_destructor : {false, true}) { + std::atomic received{false}; + std::chrono::steady_clock::time_point start; + { + sse::SSEClient sse(client, "/idle-stream"); + sse.on_message([&](const sse::SSEMessage &) { received.store(true); }); + sse.start_async(); + + for (int i = 0; i < 500 && !received.load(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + ASSERT_TRUE(received.load()); + // Let the client block in the next read + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + start = std::chrono::steady_clock::now(); + if (!by_destructor) { sse.stop(); } + } + EXPECT_LT(std::chrono::steady_clock::now() - start, + std::chrono::seconds(2)); + } +} + +TEST_F(SSEIntegrationTest, StopInterruptsBlockedRead) { + server_->Get("/idle-stream", idle_event_stream(stop_server_)); + + Client client("localhost", get_port()); + check_stop_interrupts_blocked_read(client); +} + +#ifdef CPPHTTPLIB_SSL_ENABLED +TEST(SSEClientSSLTest, StopInterruptsBlockedRead) { + SSLServer svr(SERVER_CERT_FILE, SERVER_PRIVATE_KEY_FILE); + std::atomic done{false}; + svr.Get("/idle-stream", idle_event_stream(done)); + + auto port = svr.bind_to_any_port(HOST); + std::thread t([&] { svr.listen_after_bind(); }); + auto se = detail::scope_exit([&] { + done.store(true); + svr.stop(); + t.join(); + }); + svr.wait_until_ready(); + + Client client("https://localhost:" + std::to_string(port)); + client.enable_server_certificate_verification(false); + check_stop_interrupts_blocked_read(client); +} +#endif + // Test: SSEClient with custom headers TEST_F(SSEIntegrationTest, CustomHeaders) { // Setup a server endpoint that checks for custom header