diff --git a/README-sse.md b/README-sse.md index ebfbb093..f5bb548f 100644 --- a/README-sse.md +++ b/README-sse.md @@ -79,7 +79,7 @@ sse.set_max_reconnect_attempts(10); sse.set_headers({{"Authorization", "Bearer new_token"}}); ``` -An event, or a line still waiting for its end, longer than `CPPHTTPLIB_SSE_MAX_EVENT_LENGTH` (16MB by default, `0` for no limit) ends the connection with `Error::ExceedMaxPayloadSize`, and the client stops rather than reconnect to the same event. The whole stream also counts against the client's `set_payload_max_length()` (100MB by default), so set it to `0` for a long-lived stream. +An event, or a line still waiting for its end, longer than `CPPHTTPLIB_SSE_MAX_EVENT_LENGTH` (16MB by default, `0` for no limit) ends the connection with `Error::ExceedMaxPayloadSize`, and the client stops rather than reconnect to the same event. The client's `set_payload_max_length()` does not apply to the stream as a whole, which has no end. #### Control diff --git a/httplib.h b/httplib.h index c3a85a98..b0ccd761 100644 --- a/httplib.h +++ b/httplib.h @@ -593,6 +593,10 @@ namespace ws { class WebSocket; } // namespace ws +namespace sse { +class SSEClient; +} // namespace sse + namespace detail { /* @@ -2752,6 +2756,7 @@ public: private: friend class ClientImpl; + friend class sse::SSEClient; ssize_t read_with_decompression(char *buf, size_t len); @@ -5020,8 +5025,11 @@ inline void SSEClient::run_event_loop() { request_headers.emplace("Last-Event-ID", last_event_id_); } - // Open streaming connection - auto result = stream::Get(client_, path_, request_headers); + // Open streaming connection. An event stream has no end, so each event is + // bounded instead of the whole body. + auto handle = client_.open_stream("GET", path_, {}, request_headers); + handle.body_reader_.payload_max_length = 0; + stream::Result result{std::move(handle)}; // Connection error handling if (!result) { diff --git a/test/test.cc b/test/test.cc index 56f89c2e..b5030df1 100644 --- a/test/test.cc +++ b/test/test.cc @@ -24623,9 +24623,6 @@ TEST_F(SSEIntegrationTest, TooLongEventEndsTheConnection) { }); Client client("localhost", get_port()); - // Lift the payload limit, so that only the event limit can end this endless - // stream. - client.set_payload_max_length(0); sse::SSEClient sse(client, "/endless-line"); std::atomic too_long{false}; sse.on_error([&](Error err) { @@ -24641,6 +24638,42 @@ TEST_F(SSEIntegrationTest, TooLongEventEndsTheConnection) { EXPECT_TRUE(too_long); } +TEST_F(SSEIntegrationTest, PayloadLimitDoesNotEndTheStream) { + // Far more in all than the client's payload limit below. + server_->Get("/many-events", [](const Request &, Response &res) { + res.set_chunked_content_provider( + "text/event-stream", [](size_t offset, DataSink &sink) { + if (offset == 0) { + std::string events; + for (int i = 0; i < 1000; i++) { + events += "data: message\n\n"; + } + sink.write(events.data(), events.size()); + } + return false; + }); + }); + + Client client("localhost", get_port()); + client.set_payload_max_length(1024); + sse::SSEClient sse(client, "/many-events"); + std::atomic messages{0}; + std::atomic too_large{false}; + sse.on_message([&](const sse::SSEMessage &) { messages++; }); + sse.on_error([&](Error err) { + if (err == Error::ExceedMaxPayloadSize) { too_large = true; } + }); + sse.set_max_reconnect_attempts(1); + sse.start_async(); + + for (int i = 0; i < 500 && messages < 1000; i++) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + sse.stop(); + EXPECT_GE(messages.load(), 1000); + EXPECT_FALSE(too_large); +} + // Test: on_message callback TEST_F(SSEIntegrationTest, OnMessageCallback) { // Endpoint that sends multiple events then closes