Move stream and sse implementations from the decl area to the impl area. (#2352)

This commit is contained in:
yhirose authored and GitHub committed 2026-02-09 16:41:49 -10:00
1 parent f0b7d4161d
commit a875292153
1 file changed
+208 -158
+208 -158
View File
@@ -2975,98 +2975,30 @@ namespace stream {
class Result { class Result {
public: public:
Result() : chunk_size_(8192) {} Result();
explicit Result(ClientImpl::StreamHandle &&handle, size_t chunk_size = 8192);
explicit Result(ClientImpl::StreamHandle &&handle, size_t chunk_size = 8192) Result(Result &&other) noexcept;
: handle_(std::move(handle)), chunk_size_(chunk_size) {} Result &operator=(Result &&other) noexcept;
Result(Result &&other) noexcept
: handle_(std::move(other.handle_)), buffer_(std::move(other.buffer_)),
current_size_(other.current_size_), chunk_size_(other.chunk_size_),
finished_(other.finished_) {
other.current_size_ = 0;
other.finished_ = true;
}
Result &operator=(Result &&other) noexcept {
if (this != &other) {
handle_ = std::move(other.handle_);
buffer_ = std::move(other.buffer_);
current_size_ = other.current_size_;
chunk_size_ = other.chunk_size_;
finished_ = other.finished_;
other.current_size_ = 0;
other.finished_ = true;
}
return *this;
}
Result(const Result &) = delete; Result(const Result &) = delete;
Result &operator=(const Result &) = delete; Result &operator=(const Result &) = delete;
// Check if the result is valid (connection succeeded and response received) // Response info
bool is_valid() const { return handle_.is_valid(); } bool is_valid() const;
explicit operator bool() const { return is_valid(); } explicit operator bool() const;
int status() const;
// Response status code const Headers &headers() const;
int status() const {
return handle_.response ? handle_.response->status : -1;
}
// Response headers
const Headers &headers() const {
static const Headers empty_headers;
return handle_.response ? handle_.response->headers : empty_headers;
}
std::string get_header_value(const std::string &key, std::string get_header_value(const std::string &key,
const char *def = "") const { const char *def = "") const;
return handle_.response ? handle_.response->get_header_value(key, def) bool has_header(const std::string &key) const;
: def; Error error() const;
} Error read_error() const;
bool has_read_error() const;
bool has_header(const std::string &key) const { // Stream reading
return handle_.response ? handle_.response->has_header(key) : false; bool next();
} const char *data() const;
size_t size() const;
// Error information std::string read_all();
Error error() const { return handle_.error; }
Error read_error() const { return handle_.get_read_error(); }
bool has_read_error() const { return handle_.has_read_error(); }
// Streaming iteration API
// Call next() to read the next chunk, then access data via data()/size()
// Returns true if data was read, false when stream is exhausted
bool next() {
if (!handle_.is_valid() || finished_) { return false; }
if (buffer_.size() < chunk_size_) { buffer_.resize(chunk_size_); }
ssize_t n = handle_.read(&buffer_[0], chunk_size_);
if (n > 0) {
current_size_ = static_cast<size_t>(n);
return true;
}
current_size_ = 0;
finished_ = true;
return false;
}
// Pointer to current chunk data (valid after next() returns true)
const char *data() const { return buffer_.data(); }
// Size of current chunk (valid after next() returns true)
size_t size() const { return current_size_; }
// Convenience method: read all remaining data into a string
std::string read_all() {
std::string result;
while (next()) {
result.append(data(), size());
}
return result;
}
private: private:
ClientImpl::StreamHandle handle_; ClientImpl::StreamHandle handle_;
@@ -3331,13 +3263,8 @@ struct SSEMessage {
std::string data; // Event payload std::string data; // Event payload
std::string id; // Event ID for Last-Event-ID header std::string id; // Event ID for Last-Event-ID header
SSEMessage() : event("message") {} SSEMessage();
void clear();
void clear() {
event = "message";
data.clear();
id.clear();
}
}; };
class SSEClient { class SSEClient {
@@ -3346,75 +3273,232 @@ public:
using ErrorHandler = std::function<void(Error)>; using ErrorHandler = std::function<void(Error)>;
using OpenHandler = std::function<void()>; using OpenHandler = std::function<void()>;
SSEClient(Client &client, const std::string &path) SSEClient(Client &client, const std::string &path);
: client_(client), path_(path) {} SSEClient(Client &client, const std::string &path, const Headers &headers);
~SSEClient();
SSEClient(Client &client, const std::string &path, const Headers &headers)
: client_(client), path_(path), headers_(headers) {}
~SSEClient() { stop(); }
SSEClient(const SSEClient &) = delete; SSEClient(const SSEClient &) = delete;
SSEClient &operator=(const SSEClient &) = delete; SSEClient &operator=(const SSEClient &) = delete;
// Event handlers // Event handlers
SSEClient &on_message(MessageHandler handler) { SSEClient &on_message(MessageHandler handler);
SSEClient &on_event(const std::string &type, MessageHandler handler);
SSEClient &on_open(OpenHandler handler);
SSEClient &on_error(ErrorHandler handler);
SSEClient &set_reconnect_interval(int ms);
SSEClient &set_max_reconnect_attempts(int n);
// State accessors
bool is_connected() const;
const std::string &last_event_id() const;
// Blocking start - runs event loop with auto-reconnect
void start();
// Non-blocking start - runs in background thread
void start_async();
// Stop the client (thread-safe)
void stop();
private:
bool parse_sse_line(const std::string &line, SSEMessage &msg, int &retry_ms);
void run_event_loop();
void dispatch_event(const SSEMessage &msg);
bool should_reconnect(int count) const;
void wait_for_reconnect();
// Client and path
Client &client_;
std::string path_;
Headers headers_;
// Callbacks
MessageHandler on_message_;
std::map<std::string, MessageHandler> event_handlers_;
OpenHandler on_open_;
ErrorHandler on_error_;
// Configuration
int reconnect_interval_ms_ = 3000;
int max_reconnect_attempts_ = 0; // 0 = unlimited
// State
std::atomic<bool> running_{false};
std::atomic<bool> connected_{false};
std::string last_event_id_;
// Async support
std::thread async_thread_;
};
} // namespace sse
// ----------------------------------------------------------------------------
/*
* Implementation that will be part of the .cc file if split into .h + .cc.
*/
namespace stream {
// stream::Result implementations
inline Result::Result() : chunk_size_(8192) {}
inline Result::Result(ClientImpl::StreamHandle &&handle, size_t chunk_size)
: handle_(std::move(handle)), chunk_size_(chunk_size) {}
inline Result::Result(Result &&other) noexcept
: handle_(std::move(other.handle_)), buffer_(std::move(other.buffer_)),
current_size_(other.current_size_), chunk_size_(other.chunk_size_),
finished_(other.finished_) {
other.current_size_ = 0;
other.finished_ = true;
}
inline Result &Result::operator=(Result &&other) noexcept {
if (this != &other) {
handle_ = std::move(other.handle_);
buffer_ = std::move(other.buffer_);
current_size_ = other.current_size_;
chunk_size_ = other.chunk_size_;
finished_ = other.finished_;
other.current_size_ = 0;
other.finished_ = true;
}
return *this;
}
inline bool Result::is_valid() const { return handle_.is_valid(); }
inline Result::operator bool() const { return is_valid(); }
inline int Result::status() const {
return handle_.response ? handle_.response->status : -1;
}
inline const Headers &Result::headers() const {
static const Headers empty_headers;
return handle_.response ? handle_.response->headers : empty_headers;
}
inline std::string Result::get_header_value(const std::string &key,
const char *def) const {
return handle_.response ? handle_.response->get_header_value(key, def) : def;
}
inline bool Result::has_header(const std::string &key) const {
return handle_.response ? handle_.response->has_header(key) : false;
}
inline Error Result::error() const { return handle_.error; }
inline Error Result::read_error() const { return handle_.get_read_error(); }
inline bool Result::has_read_error() const { return handle_.has_read_error(); }
inline bool Result::next() {
if (!handle_.is_valid() || finished_) { return false; }
if (buffer_.size() < chunk_size_) { buffer_.resize(chunk_size_); }
ssize_t n = handle_.read(&buffer_[0], chunk_size_);
if (n > 0) {
current_size_ = static_cast<size_t>(n);
return true;
}
current_size_ = 0;
finished_ = true;
return false;
}
inline const char *Result::data() const { return buffer_.data(); }
inline size_t Result::size() const { return current_size_; }
inline std::string Result::read_all() {
std::string result;
while (next()) {
result.append(data(), size());
}
return result;
}
} // namespace stream
namespace sse {
// SSEMessage implementations
inline SSEMessage::SSEMessage() : event("message") {}
inline void SSEMessage::clear() {
event = "message";
data.clear();
id.clear();
}
// SSEClient implementations
inline SSEClient::SSEClient(Client &client, const std::string &path)
: client_(client), path_(path) {}
inline SSEClient::SSEClient(Client &client, const std::string &path,
const Headers &headers)
: client_(client), path_(path), headers_(headers) {}
inline SSEClient::~SSEClient() { stop(); }
inline SSEClient &SSEClient::on_message(MessageHandler handler) {
on_message_ = std::move(handler); on_message_ = std::move(handler);
return *this; return *this;
} }
SSEClient &on_event(const std::string &type, MessageHandler handler) { inline SSEClient &SSEClient::on_event(const std::string &type,
MessageHandler handler) {
event_handlers_[type] = std::move(handler); event_handlers_[type] = std::move(handler);
return *this; return *this;
} }
SSEClient &on_open(OpenHandler handler) { inline SSEClient &SSEClient::on_open(OpenHandler handler) {
on_open_ = std::move(handler); on_open_ = std::move(handler);
return *this; return *this;
} }
SSEClient &on_error(ErrorHandler handler) { inline SSEClient &SSEClient::on_error(ErrorHandler handler) {
on_error_ = std::move(handler); on_error_ = std::move(handler);
return *this; return *this;
} }
SSEClient &set_reconnect_interval(int ms) { inline SSEClient &SSEClient::set_reconnect_interval(int ms) {
reconnect_interval_ms_ = ms; reconnect_interval_ms_ = ms;
return *this; return *this;
} }
SSEClient &set_max_reconnect_attempts(int n) { inline SSEClient &SSEClient::set_max_reconnect_attempts(int n) {
max_reconnect_attempts_ = n; max_reconnect_attempts_ = n;
return *this; return *this;
} }
// State accessors inline bool SSEClient::is_connected() const { return connected_.load(); }
bool is_connected() const { return connected_.load(); }
const std::string &last_event_id() const { return last_event_id_; }
// Blocking start - runs event loop with auto-reconnect inline const std::string &SSEClient::last_event_id() const {
void start() { return last_event_id_;
}
inline void SSEClient::start() {
running_.store(true); running_.store(true);
run_event_loop(); run_event_loop();
} }
// Non-blocking start - runs in background thread inline void SSEClient::start_async() {
void start_async() {
running_.store(true); running_.store(true);
async_thread_ = std::thread([this]() { run_event_loop(); }); async_thread_ = std::thread([this]() { run_event_loop(); });
} }
// Stop the client (thread-safe) inline void SSEClient::stop() {
void stop() {
running_.store(false); running_.store(false);
client_.stop(); // Cancel any pending operations client_.stop(); // Cancel any pending operations
if (async_thread_.joinable()) { async_thread_.join(); } if (async_thread_.joinable()) { async_thread_.join(); }
} }
private: inline bool SSEClient::parse_sse_line(const std::string &line, SSEMessage &msg,
// Parse a single SSE field line int &retry_ms) {
// Returns true if this line ends an event (blank line)
bool parse_sse_line(const std::string &line, SSEMessage &msg, int &retry_ms) {
// Blank line signals end of event // Blank line signals end of event
if (line.empty() || line == "\r") { return true; } if (line.empty() || line == "\r") { return true; }
@@ -3464,8 +3548,7 @@ private:
return false; return false;
} }
// Main event loop with auto-reconnect inline void SSEClient::run_event_loop() {
void run_event_loop() {
auto reconnect_count = 0; auto reconnect_count = 0;
while (running_.load()) { while (running_.load()) {
@@ -3566,8 +3649,7 @@ private:
connected_.store(false); connected_.store(false);
} }
// Dispatch event to appropriate handler inline void SSEClient::dispatch_event(const SSEMessage &msg) {
void dispatch_event(const SSEMessage &msg) {
// Check for specific event type handler first // Check for specific event type handler first
auto it = event_handlers_.find(msg.event); auto it = event_handlers_.find(msg.event);
if (it != event_handlers_.end()) { if (it != event_handlers_.end()) {
@@ -3579,15 +3661,13 @@ private:
if (on_message_) { on_message_(msg); } if (on_message_) { on_message_(msg); }
} }
// Check if we should attempt to reconnect inline bool SSEClient::should_reconnect(int count) const {
bool should_reconnect(int count) const {
if (!running_.load()) { return false; } if (!running_.load()) { return false; }
if (max_reconnect_attempts_ == 0) { return true; } // unlimited if (max_reconnect_attempts_ == 0) { return true; } // unlimited
return count < max_reconnect_attempts_; return count < max_reconnect_attempts_;
} }
// Wait for reconnect interval inline void SSEClient::wait_for_reconnect() {
void wait_for_reconnect() {
// Use small increments to check running_ flag frequently // Use small increments to check running_ flag frequently
auto waited = 0; auto waited = 0;
while (running_.load() && waited < reconnect_interval_ms_) { while (running_.load() && waited < reconnect_interval_ms_) {
@@ -3596,38 +3676,8 @@ private:
} }
} }
// Client and path
Client &client_;
std::string path_;
Headers headers_;
// Callbacks
MessageHandler on_message_;
std::map<std::string, MessageHandler> event_handlers_;
OpenHandler on_open_;
ErrorHandler on_error_;
// Configuration
int reconnect_interval_ms_ = 3000;
int max_reconnect_attempts_ = 0; // 0 = unlimited
// State
std::atomic<bool> running_{false};
std::atomic<bool> connected_{false};
std::string last_event_id_;
// Async support
std::thread async_thread_;
};
} // namespace sse } // namespace sse
// ----------------------------------------------------------------------------
/*
* Implementation that will be part of the .cc file if split into .h + .cc.
*/
#ifdef CPPHTTPLIB_SSL_ENABLED #ifdef CPPHTTPLIB_SSL_ENABLED
/* /*
* TLS abstraction layer - internal function declarations * TLS abstraction layer - internal function declarations