diff --git a/httplib.h b/httplib.h index 6ad8d1b5..2c709d52 100644 --- a/httplib.h +++ b/httplib.h @@ -2020,6 +2020,12 @@ public: // each one. 0 lifts the bound. virtual void set_max_timeout(time_t msec) { (void)msec; } + // Lets a wait for data end early once `interrupted` returns true, so that + // another thread can stop a blocked read on every platform. + virtual void set_read_interrupt(std::function interrupted) { + (void)interrupted; + } + // Bytes already pulled off the socket and sitting in this stream's own // buffer. Exposing them lets a line reader scan for a terminator in one // pass instead of asking for a byte at a time. A stream that does no @@ -5075,6 +5081,10 @@ 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; + // stop() ends a read that is waiting for the next event + if (handle.stream_) { + handle.stream_->set_read_interrupt([this] { return !running_.load(); }); + } stream::Result result{std::move(handle)}; // Connection error handling @@ -6740,6 +6750,26 @@ inline ssize_t select_until(socket_t sock, short events, static_cast(left % 1000000)); } +// select_read(), but waits in slices and gives up once interrupted() returns +// true. A local shutdown() would not wake the wait on Windows. +inline ssize_t +select_read_interruptible(socket_t sock, time_t sec, time_t usec, + const std::function &interrupted) { + if (!interrupted) { return select_read(sock, sec, usec); } + + const auto slice = std::chrono::microseconds(100000); + auto deadline = deadline_after(sec, usec); + while (!interrupted()) { + auto left = std::chrono::duration_cast( + deadline - std::chrono::steady_clock::now()); + auto wait = + (std::max)((std::min)(left, slice), std::chrono::microseconds(0)); + auto ret = select_impl(sock, POLLIN, 0, static_cast(wait.count())); + if (ret != 0 || wait < slice) { return ret; } + } + return 0; +} + inline Error wait_until_socket_is_ready(socket_t sock, time_t sec, time_t usec) { struct pollfd pfd_read; @@ -6798,6 +6828,7 @@ public: time_t duration() const override; void set_read_timeout(time_t sec, time_t usec = 0) override; void set_max_timeout(time_t msec) override; + void set_read_interrupt(std::function interrupted) override; const char *buffered_data(size_t &size) const override; void consume_buffered(size_t size) override; @@ -6819,6 +6850,7 @@ private: time_t write_timeout_usec_; time_t max_timeout_msec_; std::chrono::time_point start_time_; + std::function read_interrupted_; std::vector read_buff_; size_t read_buff_off_ = 0; @@ -10679,6 +10711,7 @@ public: time_t duration() const override; void set_read_timeout(time_t sec, time_t usec = 0) override; void set_max_timeout(time_t msec) override; + void set_read_interrupt(std::function interrupted) override; // See SocketStream::set_readable_hint(). void set_readable_hint() { readable_hint_ = true; } @@ -10698,6 +10731,7 @@ private: time_t write_timeout_usec_; time_t max_timeout_msec_; std::chrono::time_point start_time_; + std::function read_interrupted_; bool readable_hint_ = false; tls::TlsError tls_error_; }; @@ -12610,7 +12644,8 @@ inline bool SocketStream::is_readable() const { inline bool SocketStream::wait_readable() const { if (max_timeout_msec_ <= 0) { - return select_read(sock_, read_timeout_sec_, read_timeout_usec_) > 0; + return select_read_interruptible(sock_, read_timeout_sec_, + read_timeout_usec_, read_interrupted_) > 0; } time_t read_timeout_sec; @@ -12618,7 +12653,8 @@ inline bool SocketStream::wait_readable() const { calc_actual_timeout(max_timeout_msec_, duration(), read_timeout_sec_, read_timeout_usec_, read_timeout_sec, read_timeout_usec); - return select_read(sock_, read_timeout_sec, read_timeout_usec) > 0; + return select_read_interruptible(sock_, read_timeout_sec, read_timeout_usec, + read_interrupted_) > 0; } inline bool SocketStream::wait_writable() const { @@ -12749,6 +12785,11 @@ inline void SocketStream::set_max_timeout(time_t msec) { start_time_ = std::chrono::steady_clock::now(); } +inline void +SocketStream::set_read_interrupt(std::function interrupted) { + read_interrupted_ = std::move(interrupted); +} + // Buffer stream implementation inline bool BufferStream::is_readable() const { return true; } @@ -13174,7 +13215,8 @@ inline bool SSLSocketStream::is_readable() const { inline bool SSLSocketStream::wait_readable() const { if (max_timeout_msec_ <= 0) { - return select_read(sock_, read_timeout_sec_, read_timeout_usec_) > 0; + return select_read_interruptible(sock_, read_timeout_sec_, + read_timeout_usec_, read_interrupted_) > 0; } time_t read_timeout_sec; @@ -13182,7 +13224,8 @@ inline bool SSLSocketStream::wait_readable() const { calc_actual_timeout(max_timeout_msec_, duration(), read_timeout_sec_, read_timeout_usec_, read_timeout_sec, read_timeout_usec); - return select_read(sock_, read_timeout_sec, read_timeout_usec) > 0; + return select_read_interruptible(sock_, read_timeout_sec, read_timeout_usec, + read_interrupted_) > 0; } inline bool SSLSocketStream::wait_writable() const { @@ -13309,6 +13352,11 @@ inline void SSLSocketStream::set_max_timeout(time_t msec) { start_time_ = std::chrono::steady_clock::now(); } +inline void +SSLSocketStream::set_read_interrupt(std::function interrupted) { + read_interrupted_ = std::move(interrupted); +} + inline WebSocketSSLStream::WebSocketSSLStream(socket_t sock, tls::session_t session, time_t read_timeout_sec, diff --git a/test/test.cc b/test/test.cc index 40c7dd1d..75c9c80f 100644 --- a/test/test.cc +++ b/test/test.cc @@ -25593,19 +25593,16 @@ TEST_F(SSEIntegrationTest, LastEventIdHeader) { // Test: Manual stop TEST_F(SSEIntegrationTest, ManualStop) { - // Endpoint that sends one event and stays open briefly - std::atomic handler_running{true}; - - server_->Get("/stay-open", [&handler_running](const Request &, - Response &res) { + // Endpoint that sends one event and stays open briefly. The server outlives + // this test body, so the provider uses the fixture's flag, not a local. + server_->Get("/stay-open", [this](const Request &, Response &res) { res.set_chunked_content_provider( - "text/event-stream", [&handler_running](size_t offset, DataSink &sink) { + "text/event-stream", [this](size_t offset, DataSink &sink) { if (offset == 0) { std::string event = "data: connected\n\n"; sink.write(event.data(), event.size()); } - // Keep connection open while handler_running is true - for (int i = 0; i < 10 && handler_running.load(); ++i) { + for (int i = 0; i < 10 && !stop_server_.load(); ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(50)); } return false; @@ -25630,13 +25627,85 @@ TEST_F(SSEIntegrationTest, ManualStop) { EXPECT_TRUE(sse.is_connected()); // Signal handler to stop - handler_running.store(false); + stop_server_.store(true); // Stop SSE client sse.stop(); 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