Let SSEClient::stop() interrupt a blocked read

The event loop reads through open_stream(), which takes the socket out
of the Client, so the client_.stop() in stop() could not reach it.
stop(), and the destructor with it, waited until the server sent
something or the read timeout passed, 300 seconds by default.

The loop now hands the stream's socket to stop() while it reads, and
withdraws it under the same lock before the socket is closed. stop()
shuts down the read side only: a TLS backend answers the EOF with an
alert, and writing it to a socket closed for writing raises SIGPIPE.

httplib.h

@@ -4493,6 +4493,10 @@ private:
   std::atomic<bool> 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<std::mutex> 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<std::mutex> 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<std::mutex> lock(stream_mutex_);
+      stream_sock_ = INVALID_SOCKET;
+    });
+
     // Connection successful
     connected_.store(true);
     reconnect_count = 0;

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<bool> &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<bool> 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<bool> 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