mirror of
https://github.com/yhirose/cpp-httplib.git
synced 2026-10-19 11:19:36 +00:00
Bound an SSE stream by event only
An event stream has no end, but the client's payload_max_length, 100MB by default, cut it and made the client reconnect. SSEClient lifts that limit for its stream; each event stays bounded by CPPHTTPLIB_SSE_MAX_EVENT_LENGTH.
This commit is contained in:
1 parent
35342285a0
commit
ad6c851d19
3 files changed
+47
-6
No files matched your search
+1
-1
@@ -79,7 +79,7 @@ sse.set_max_reconnect_attempts(10);
|
|||||||
sse.set_headers({{"Authorization", "Bearer new_token"}});
|
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
|
#### Control
|
||||||
|
|
||||||
|
|||||||
@@ -593,6 +593,10 @@ namespace ws {
|
|||||||
class WebSocket;
|
class WebSocket;
|
||||||
} // namespace ws
|
} // namespace ws
|
||||||
|
|
||||||
|
namespace sse {
|
||||||
|
class SSEClient;
|
||||||
|
} // namespace sse
|
||||||
|
|
||||||
namespace detail {
|
namespace detail {
|
||||||
|
|
||||||
/*
|
/*
|
||||||
@@ -2752,6 +2756,7 @@ public:
|
|||||||
|
|
||||||
private:
|
private:
|
||||||
friend class ClientImpl;
|
friend class ClientImpl;
|
||||||
|
friend class sse::SSEClient;
|
||||||
|
|
||||||
ssize_t read_with_decompression(char *buf, size_t len);
|
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_);
|
request_headers.emplace("Last-Event-ID", last_event_id_);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Open streaming connection
|
// Open streaming connection. An event stream has no end, so each event is
|
||||||
auto result = stream::Get(client_, path_, request_headers);
|
// 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
|
// Connection error handling
|
||||||
if (!result) {
|
if (!result) {
|
||||||
|
|||||||
+36
-3
@@ -24623,9 +24623,6 @@ TEST_F(SSEIntegrationTest, TooLongEventEndsTheConnection) {
|
|||||||
});
|
});
|
||||||
|
|
||||||
Client client("localhost", get_port());
|
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");
|
sse::SSEClient sse(client, "/endless-line");
|
||||||
std::atomic<bool> too_long{false};
|
std::atomic<bool> too_long{false};
|
||||||
sse.on_error([&](Error err) {
|
sse.on_error([&](Error err) {
|
||||||
@@ -24641,6 +24638,42 @@ TEST_F(SSEIntegrationTest, TooLongEventEndsTheConnection) {
|
|||||||
EXPECT_TRUE(too_long);
|
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<int> messages{0};
|
||||||
|
std::atomic<bool> 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: on_message callback
|
||||||
TEST_F(SSEIntegrationTest, OnMessageCallback) {
|
TEST_F(SSEIntegrationTest, OnMessageCallback) {
|
||||||
// Endpoint that sends multiple events then closes
|
// Endpoint that sends multiple events then closes
|
||||||
|
|||||||
Reference in new issue
Block a user