From bd1d7ca4d08362a10c1134f5ccf394f57c40fa1c Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 1 Aug 2026 13:15:52 +0800 Subject: [PATCH] refactor(libhttp): split HttpServer into modular TUs for faster incremental builds Split the 1400-line HttpServer.cpp into 6 focused translation units by functional area: - HttpServer.cpp: core lifecycle, SSL init, session management, debug - HttpServer_Listener.cpp: TCP connection acceptor - HttpServer_Routes.cpp: route registration, matching, wildcard/query parsing - HttpServer_Request.cpp: request dispatch, static files, CORS, HTTP I/O - HttpServer_Ws.cpp: WS route registry, client management, broadcast APIs - HttpServer_WsSession.cpp: per-session WS frames + bidirectional WS proxy Add inc/us/detail/HttpServerImpl.hpp for internal shared definitions (exception classes, full Session/Listener declarations) installed under /usr/include/us/detail/. Update Makefile with proper dependency tracking. Update README with Source Layout section and v3.3.0 changelog entry. Modifying one module now only requires recompiling its corresponding .o instead of the entire monolithic translation unit. --- libhttp/Makefile | 38 +- libhttp/README.md | 18 + libhttp/inc/us/detail/HttpServerImpl.hpp | 180 ++++ libhttp/src/HttpServer.cpp | 1247 +--------------------- libhttp/src/HttpServer_Listener.cpp | 51 + libhttp/src/HttpServer_Request.cpp | 329 ++++++ libhttp/src/HttpServer_Routes.cpp | 125 +++ libhttp/src/HttpServer_Ws.cpp | 106 ++ libhttp/src/HttpServer_WsSession.cpp | 506 +++++++++ 9 files changed, 1373 insertions(+), 1227 deletions(-) create mode 100644 libhttp/inc/us/detail/HttpServerImpl.hpp create mode 100644 libhttp/src/HttpServer_Listener.cpp create mode 100644 libhttp/src/HttpServer_Request.cpp create mode 100644 libhttp/src/HttpServer_Routes.cpp create mode 100644 libhttp/src/HttpServer_Ws.cpp create mode 100644 libhttp/src/HttpServer_WsSession.cpp diff --git a/libhttp/Makefile b/libhttp/Makefile index 41a1bb0..eaaa3ec 100644 --- a/libhttp/Makefile +++ b/libhttp/Makefile @@ -13,15 +13,38 @@ CLIENT_LIBS := -L/opt/homebrew/lib -lssl -lcrypto -lcurl -lpthread BUILD_TDIR := build $(shell mkdir -p $(BUILD_TDIR)) +# Server object files (HttpServer 拆分后) +SERVER_OBJS := $(BUILD_TDIR)/HttpServer.o \ + $(BUILD_TDIR)/HttpServer_Listener.o \ + $(BUILD_TDIR)/HttpServer_Routes.o \ + $(BUILD_TDIR)/HttpServer_Request.o \ + $(BUILD_TDIR)/HttpServer_Ws.o \ + $(BUILD_TDIR)/HttpServer_WsSession.o + #-------- all: $(BUILD_TDIR)/$(LIB_NAME).so $(BUILD_TDIR)/$(LIBC_NAME).so # ---- Server library (HttpServer, no curl) ---- -$(BUILD_TDIR)/$(LIB_NAME).so: $(BUILD_TDIR)/HttpServer.o +$(BUILD_TDIR)/$(LIB_NAME).so: $(SERVER_OBJS) $(CPP) -shared $^ $(SERVER_LIBS) -o $@ -$(BUILD_TDIR)/HttpServer.o: src/HttpServer.cpp - $(CPP) $(CPPFLAGS) -fPIC -c $^ $(INCS) -o $@ +$(BUILD_TDIR)/HttpServer.o: src/HttpServer.cpp inc/us/detail/HttpServerImpl.hpp + $(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@ + +$(BUILD_TDIR)/HttpServer_Listener.o: src/HttpServer_Listener.cpp inc/us/detail/HttpServerImpl.hpp + $(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@ + +$(BUILD_TDIR)/HttpServer_Routes.o: src/HttpServer_Routes.cpp inc/us/detail/HttpServerImpl.hpp + $(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@ + +$(BUILD_TDIR)/HttpServer_Request.o: src/HttpServer_Request.cpp inc/us/detail/HttpServerImpl.hpp + $(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@ + +$(BUILD_TDIR)/HttpServer_Ws.o: src/HttpServer_Ws.cpp inc/us/detail/HttpServerImpl.hpp + $(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@ + +$(BUILD_TDIR)/HttpServer_WsSession.o: src/HttpServer_WsSession.cpp inc/us/detail/HttpServerImpl.hpp + $(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@ # ---- Client library (HttpClient, depends on curl + beast websocket) ---- $(BUILD_TDIR)/$(LIBC_NAME).so: $(BUILD_TDIR)/HttpClient.o @@ -36,8 +59,13 @@ clear: install: all @echo "install UniServe_libhttp..." @mkdir -vp $(INCLU_TDIR)/$(FRAMEWORK_NAME) + @mkdir -vp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/detail @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/HttpServer.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/HttpClient.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ + @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/WebSocketServer.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ + @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/WebSocketClient.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ + @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/WebSocketClientBase.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ + @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/detail/HttpServerImpl.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/detail/ @cp -vf $(BUILD_TDIR)/$(LIB_NAME).so $(INSTL_TDIR)/ @cp -vf $(BUILD_TDIR)/$(LIBC_NAME).so $(INSTL_TDIR)/ @echo "done." @@ -46,6 +74,10 @@ uninstall: @echo "uninstall UniServe_libhttp..." @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/HttpServer.hpp @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/HttpClient.hpp + @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/WebSocketServer.hpp + @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/WebSocketClient.hpp + @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/WebSocketClientBase.hpp + @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/detail/HttpServerImpl.hpp @rm -vf $(INSTL_TDIR)/$(LIB_NAME).so @rm -vf $(INSTL_TDIR)/$(LIBC_NAME).so @echo "done." diff --git a/libhttp/README.md b/libhttp/README.md index 4813da3..f96a371 100644 --- a/libhttp/README.md +++ b/libhttp/README.md @@ -46,6 +46,23 @@ Two shared libraries are produced: `libus_http.so` does **not** depend on libcurl. Link against `libus_httpclient.so` only when you need the client, and link both when you need both server and client (e.g. a proxy/gateway). +### Source Layout + +The server implementation is split into modular source files for faster incremental builds: + +| Source File | Responsibility | +|---|---| +| `src/HttpServer.cpp` | Core lifecycle (construct/destruct/start/stop/run), SSL init, session management, debug utilities | +| `src/HttpServer_Listener.cpp` | TCP connection acceptor (`Listener` class) | +| `src/HttpServer_Routes.cpp` | HTTP route registration, route matching, wildcard→regex conversion, query parameter parsing | +| `src/HttpServer_Request.cpp` | Request dispatch pipeline, static file serving, CORS preflight, Session HTTP read/write | +| `src/HttpServer_Ws.cpp` | WebSocket route registration, client registry, broadcast/send APIs | +| `src/HttpServer_WsSession.cpp` | Session WebSocket handling (regular WS frames + bidirectional WS proxy bridging) | + +Internal shared definitions (exception classes, full `Session`/`Listener` declarations) live in `inc/us/detail/HttpServerImpl.hpp` (installed to `/usr/include/us/detail/`). + +Modifying one module only requires recompiling its corresponding `.o` file — no need to rebuild the entire 1400-line translation unit. + ## Quick Start ### HTTP + WebSocket Server @@ -260,6 +277,7 @@ using WebSocketClientBase = HttpClient; ## Version History +- **v3.3.0**: Split HttpServer into 6 modular translation units by functional area (core/lifecycle, listener, routes, request dispatch, WS management, WS session/proxy) for dramatically faster incremental builds. Added internal detail header `inc/us/detail/HttpServerImpl.hpp` for shared definitions across TU boundaries. - **v3.2.0**: Split into two shared libraries — `libus_http.so` (server only, no curl) and `libus_httpclient.so` (HttpClient, depends on curl). HttpClient implementation moved to `src/HttpClient.cpp`; `HttpClient.hpp` is declaration-only. - **v3.1.0**: Added WebSocket reverse proxy (`ws_proxy()`) with bidirectional async bridging, ws:// and wss:// backend support, handshake-time authorization callback. - **v3.0.0**: Merged WebSocketServer into HttpServer and WebSocketClient into HttpClient. Single-port HTTP+WebSocket support. diff --git a/libhttp/inc/us/detail/HttpServerImpl.hpp b/libhttp/inc/us/detail/HttpServerImpl.hpp new file mode 100644 index 0000000..16393b0 --- /dev/null +++ b/libhttp/inc/us/detail/HttpServerImpl.hpp @@ -0,0 +1,180 @@ +#ifndef HTTPSERVER_IMPL_HPP +#define HTTPSERVER_IMPL_HPP + +#include "us/HttpServer.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace pmc { +namespace net { + +// ===================== 自定义异常 ===================== +class HttpServerException : public std::runtime_error { +public: + explicit HttpServerException(const std::string& msg) : runtime_error(msg) {} +}; +class HttpListenerException : public HttpServerException { +public: + explicit HttpListenerException(const std::string& msg) : HttpServerException(msg) {} +}; +class SSLInitException : public HttpServerException { +public: + explicit SSLInitException(const std::string& msg) : HttpServerException(msg) {} +}; +class CreateObjectFailed : public HttpServerException { +public: + explicit CreateObjectFailed(const std::string& msg) : HttpServerException(msg) {} +}; + +// ===================== Session (完整定义) ===================== +class HttpServer::Session : public std::enable_shared_from_this { + using tcp_stream = beast::tcp_stream; + using ssl_stream = asio::ssl::stream; + using ws_stream = websocket::stream; + using wss_stream = websocket::stream; + + // Proxy outbound types + using proxy_ws_stream = websocket::stream; + using proxy_wss_stream = websocket::stream&>; + +public: + Session(tcp::socket&& socket, HttpServer* server) + : stream_(std::move(socket)), server_(server), ssl_enabled_(false), + is_websocket_(false), is_ws_proxy_(false), client_id_("") {} + + Session(tcp::socket&& socket, asio::ssl::context& ssl_ctx, HttpServer* server) + : stream_(std::move(socket)), server_(server), ssl_enabled_(true), + is_websocket_(false), is_ws_proxy_(false), client_id_("") { + ssl_stream_ = std::make_unique(stream_, ssl_ctx); + } + + ~Session() { close(); } + + void run(); + void setRemoteInfo(const std::string& ip, unsigned short port); + void wsSend(const std::string& message); + std::string getClientId() const { return client_id_; } + std::string getRemoteAddress() const { return remote_ip_ + ":" + std::to_string(remote_port_); } + bool isWebSocket() const { return is_websocket_; } + +private: + void close(); + void doSSLHandshake(); + void doRead(); + void handleRead(beast::error_code ec); + bool isWebSocketUpgrade(const http::request& req); + void doWsUpgrade(); + void doWssUpgrade(); + void onWsAccept(beast::error_code ec); + + // Regular WebSocket + void doWsRead(); + void onWsRead(beast::error_code ec, std::size_t bytes_transferred); + void doWsWrite(); + void onWsClose(); + + // WebSocket Proxy + static void parseProxyUrl(const std::string& url, bool& ssl, std::string& host, + std::string& port, std::string& path); + void doProxyConnect(const std::string& target_url); + void onProxyResolve(beast::error_code ec, tcp::resolver::results_type results); + void onProxyConnect(beast::error_code ec, tcp::endpoint ep); + void onProxySslHandshake(beast::error_code ec); + void onProxyHandshake(beast::error_code ec); + + // Client → Backend + void doProxyReadClient(); + void onProxyReadClient(beast::error_code ec, std::size_t bytes_transferred); + void doProxyWriteBackend(); + + // Backend → Client + void doProxyReadBackend(); + void onProxyReadBackend(beast::error_code ec, std::size_t bytes_transferred); + void doProxyWriteClient(); + + void closeProxy(websocket::close_code code); + + // Regular HTTP + void processRequest(http::request req); + void doWrite(std::shared_ptr> response); + + // Client stream (inbound) + tcp_stream stream_; + std::unique_ptr ssl_stream_; + std::unique_ptr ws_stream_; + std::unique_ptr wss_stream_; + + beast::flat_buffer buffer_; + beast::flat_buffer ws_buffer_; + http::request request_; + + HttpServer* server_; + bool ssl_enabled_; + + std::string remote_ip_ = "unknown"; + unsigned short remote_port_ = 0; + + bool is_websocket_; + bool is_ws_proxy_; + std::string client_id_; + std::string ws_path_; + WebSocketMessageHandler ws_handler_; + std::deque ws_send_queue_; + + // Proxy state (outbound to backend) + std::string ws_proxy_target_; + bool proxy_ssl_ = false; + bool proxy_connected_ = false; + bool proxy_closed_ = false; + std::string proxy_host_, proxy_port_, proxy_path_; + + std::unique_ptr proxy_resolver_; + std::unique_ptr proxy_socket_; + std::unique_ptr proxy_tcp_stream_; + std::unique_ptr proxy_ssl_ctx_; + std::unique_ptr> proxy_ssl_stream_; + std::unique_ptr proxy_ws_stream_; + std::unique_ptr proxy_wss_stream_; + + beast::flat_buffer proxy_client_buffer_; + beast::flat_buffer proxy_backend_buffer_; + std::deque> proxy_to_backend_queue_; + std::deque> proxy_to_client_queue_; +}; + +// ===================== Listener (完整定义) ===================== +class HttpServer::Listener : public std::enable_shared_from_this { +public: + Listener(asio::io_context& ioc, tcp::endpoint endpoint, HttpServer* server) + : ioc_(ioc), acceptor_(ioc), server_(server), ssl_enabled_(false) { + initAcceptor(endpoint); + } + + Listener(asio::io_context& ioc, tcp::endpoint endpoint, + asio::ssl::context& ssl_ctx, HttpServer* server) + : ioc_(ioc), acceptor_(ioc), ssl_ctx_(&ssl_ctx), server_(server), ssl_enabled_(true) { + initAcceptor(endpoint); + } + + void run(); + +private: + void initAcceptor(tcp::endpoint ep); + void doAccept(); + + asio::io_context& ioc_; + tcp::acceptor acceptor_; + asio::ssl::context* ssl_ctx_ = nullptr; + HttpServer* server_; + bool ssl_enabled_; +}; + +} // namespace net +} // namespace pmc + +#endif // HTTPSERVER_IMPL_HPP diff --git a/libhttp/src/HttpServer.cpp b/libhttp/src/HttpServer.cpp index aaa9580..e16811e 100644 --- a/libhttp/src/HttpServer.cpp +++ b/libhttp/src/HttpServer.cpp @@ -1,860 +1,9 @@ -#include "us/HttpServer.hpp" -#include -#include -#include -#include -#include -#include -#include +#include "us/detail/HttpServerImpl.hpp" namespace pmc { namespace net { -// ===================== 自定义异常 ===================== -class HttpServerException : public std::runtime_error { -public: - explicit HttpServerException(const std::string& msg) : runtime_error(msg) {} -}; -class HttpListenerException : public HttpServerException { -public: - explicit HttpListenerException(const std::string& msg) : HttpServerException(msg) {} -}; -class SSLInitException : public HttpServerException { -public: - explicit SSLInitException(const std::string& msg) : HttpServerException(msg) {} -}; -class CreateObjectFailed : public HttpServerException { -public: - explicit CreateObjectFailed(const std::string& msg) : HttpServerException(msg) {} -}; - -// ===================== Session ===================== -class HttpServer::Session : public std::enable_shared_from_this { - using tcp_stream = beast::tcp_stream; - using ssl_stream = asio::ssl::stream; - using ws_stream = websocket::stream; - using wss_stream = websocket::stream; - - // Proxy outbound types - using proxy_ws_stream = websocket::stream; - using proxy_wss_stream = websocket::stream&>; - -public: - Session(tcp::socket&& socket, HttpServer* server) - : stream_(std::move(socket)), server_(server), ssl_enabled_(false), - is_websocket_(false), is_ws_proxy_(false), client_id_("") {} - - Session(tcp::socket&& socket, asio::ssl::context& ssl_ctx, HttpServer* server) - : stream_(std::move(socket)), server_(server), ssl_enabled_(true), - is_websocket_(false), is_ws_proxy_(false), client_id_("") { - ssl_stream_ = std::make_unique(stream_, ssl_ctx); - } - - ~Session() { close(); } - - void run() { - if (ssl_enabled_) { - doSSLHandshake(); - } else { - doRead(); - } - } - - void setRemoteInfo(const std::string& ip, unsigned short port) { - remote_ip_ = ip; - remote_port_ = port; - } - - void wsSend(const std::string& message) { - auto exec = stream_.get_executor(); - asio::post(exec, [self = shared_from_this(), message]() { - self->ws_send_queue_.push_back(message); - if (self->ws_send_queue_.size() == 1) { - self->doWsWrite(); - } - }); - } - - std::string getClientId() const { return client_id_; } - std::string getRemoteAddress() const { return remote_ip_ + ":" + std::to_string(remote_port_); } - bool isWebSocket() const { return is_websocket_; } - -private: - void close() { - beast::error_code ec; - closeProxy(websocket::close_code::going_away); - if (is_websocket_) { - if (ssl_enabled_ && wss_stream_) { - wss_stream_->close(websocket::close_code::normal, ec); - } else if (ws_stream_) { - ws_stream_->close(websocket::close_code::normal, ec); - } - } - if (ssl_stream_) { - ssl_stream_->shutdown(ec); - } - stream_.socket().shutdown(tcp::socket::shutdown_send, ec); - stream_.close(); - } - - void doSSLHandshake() { - auto self = shared_from_this(); - std::cout << "[SSL] Starting handshake for " - << remote_ip_ << ":" << remote_port_ << std::endl; - - ssl_stream_->async_handshake( - asio::ssl::stream_base::server, - [self](beast::error_code ec) { - if (ec) { - std::cerr << "[SSL] Handshake error: " - << self->remote_ip_ << ":" << self->remote_port_ - << " - " << ec.message() - << " (code=" << ec.value() << ")" << std::endl; - ERR_print_errors_fp(stderr); - return; - } - std::cout << "[SSL] Handshake succeeded: " - << self->remote_ip_ << ":" << self->remote_port_ << std::endl; - self->doRead(); - }); - } - - void doRead() { - auto self = shared_from_this(); - request_ = {}; - buffer_.consume(buffer_.size()); - - if (ssl_enabled_ && ssl_stream_) { - http::async_read(*ssl_stream_, buffer_, request_, - [self](beast::error_code ec, std::size_t) { - self->handleRead(ec); - }); - } else { - http::async_read(stream_, buffer_, request_, - [self](beast::error_code ec, std::size_t) { - self->handleRead(ec); - }); - } - } - - void handleRead(beast::error_code ec) { - if (!ec) { - if (isWebSocketUpgrade(request_)) { - std::string path = std::string(request_.target()); - auto qpos = path.find('?'); - if (qpos != std::string::npos) path = path.substr(0, qpos); - - // Check proxy routes first (higher priority) - auto proxy_handshake = server_->findWsProxyRoute(path); - if (proxy_handshake) { - std::string target = proxy_handshake(request_); - if (target.empty()) { - // Rejected by handshake callback → 403 - auto resp = std::make_shared>(); - resp->version(request_.version()); - resp->result(http::status::forbidden); - resp->set(http::field::content_type, "text/plain"); - resp->set(http::field::connection, "close"); - resp->body() = "403 Forbidden\n"; - resp->prepare_payload(); - doWrite(resp); - return; - } - ws_path_ = path; - is_ws_proxy_ = true; - ws_proxy_target_ = target; - if (ssl_enabled_) { - doWssUpgrade(); - } else { - doWsUpgrade(); - } - return; - } - - auto ws_handler = server_->findWsRoute(path); - if (ws_handler) { - ws_handler_ = ws_handler; - ws_path_ = path; - if (ssl_enabled_) { - doWssUpgrade(); - } else { - doWsUpgrade(); - } - return; - } - } - - auto req_copy = std::move(request_); - processRequest(std::move(req_copy)); - } else if (ec != http::error::end_of_stream) { - std::cerr << "[Read] Error from " << remote_ip_ << ":" << remote_port_ - << ": " << ec.message() << std::endl; - } - } - - bool isWebSocketUpgrade(const http::request& req) { - if (req.method() != http::verb::get) return false; - - auto upgrade = req.find(http::field::upgrade); - if (upgrade == req.end()) return false; - - std::string upgrade_val(upgrade->value()); - boost::to_lower(upgrade_val); - boost::trim(upgrade_val); - if (upgrade_val != "websocket") return false; - - auto connection = req.find(http::field::connection); - if (connection == req.end()) return false; - - std::string conn_val(connection->value()); - boost::to_lower(conn_val); - return conn_val.find("upgrade") != std::string::npos; - } - - void doWsUpgrade() { - auto self = shared_from_this(); - ws_stream_ = std::make_unique(stream_); - ws_stream_->text(true); - - ws_stream_->async_accept( - request_, - [self](beast::error_code ec) { - self->onWsAccept(ec); - }); - } - - void doWssUpgrade() { - auto self = shared_from_this(); - wss_stream_ = std::make_unique(*ssl_stream_); - wss_stream_->text(true); - - wss_stream_->async_accept( - request_, - [self](beast::error_code ec) { - self->onWsAccept(ec); - }); - } - - void onWsAccept(beast::error_code ec) { - if (ec) { - std::cerr << "[WS] Accept error from " << remote_ip_ << ":" << remote_port_ - << ": " << ec.message() << std::endl; - return; - } - - is_websocket_ = true; - client_id_ = server_->generateWsClientId(); - server_->addWsClient(client_id_, shared_from_this()); - - std::cout << "[WS] Client " << client_id_ << " connected from " - << remote_ip_ << ":" << remote_port_ << " at " << ws_path_ << std::endl; - - if (is_ws_proxy_) { - std::cout << "[WS-Proxy] Proxying " << ws_path_ << " -> " << ws_proxy_target_ << std::endl; - doProxyConnect(ws_proxy_target_); - return; - } - - if (server_->ws_connect_handler_) { - server_->ws_connect_handler_(client_id_, getRemoteAddress()); - } - - doWsRead(); - } - - // ======== Regular WebSocket message handling ======== - - void doWsRead() { - auto self = shared_from_this(); - ws_buffer_.consume(ws_buffer_.size()); - - if (ssl_enabled_ && wss_stream_) { - wss_stream_->async_read( - ws_buffer_, - [self](beast::error_code ec, std::size_t bytes_transferred) { - self->onWsRead(ec, bytes_transferred); - }); - } else if (ws_stream_) { - ws_stream_->async_read( - ws_buffer_, - [self](beast::error_code ec, std::size_t bytes_transferred) { - self->onWsRead(ec, bytes_transferred); - }); - } - } - - void onWsRead(beast::error_code ec, std::size_t bytes_transferred) { - if (ec == websocket::error::closed) { - onWsClose(); - return; - } - if (ec) { - std::cerr << "[WS] Read error from " << client_id_ << ": " << ec.message() << std::endl; - onWsClose(); - return; - } - - std::string message = beast::buffers_to_string(ws_buffer_.data()); - ws_buffer_.consume(bytes_transferred); - - if (ws_handler_) { - auto self = shared_from_this(); - ws_handler_(client_id_, message, - [self](const std::string& response) { - self->wsSend(response); - }); - } - - doWsRead(); - } - - void doWsWrite() { - if (ws_send_queue_.empty()) return; - - auto self = shared_from_this(); - std::string& message = ws_send_queue_.front(); - - auto write_handler = [self](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[WS] Write error to " << self->client_id_ << ": " << ec.message() << std::endl; - self->onWsClose(); - return; - } - self->ws_send_queue_.pop_front(); - if (!self->ws_send_queue_.empty()) { - self->doWsWrite(); - } - }; - - if (ssl_enabled_ && wss_stream_) { - wss_stream_->text(true); - wss_stream_->async_write(asio::buffer(message), write_handler); - } else if (ws_stream_) { - ws_stream_->text(true); - ws_stream_->async_write(asio::buffer(message), write_handler); - } - } - - void onWsClose() { - if (client_id_.empty()) return; - std::cout << "[WS] Client " << client_id_ << " disconnected" << std::endl; - - closeProxy(websocket::close_code::normal); - server_->removeWsClient(client_id_); - - if (server_->ws_disconnect_handler_) { - server_->ws_disconnect_handler_(client_id_); - } - client_id_.clear(); - is_websocket_ = false; - is_ws_proxy_ = false; - } - - // ======== WebSocket Proxy (bidirectional bridge) ======== - - static void parseProxyUrl(const std::string& url, bool& ssl, std::string& host, - std::string& port, std::string& path) { - std::string rest; - if (url.size() >= 6 && url.substr(0, 6) == "wss://") { - ssl = true; - rest = url.substr(6); - } else if (url.size() >= 5 && url.substr(0, 5) == "ws://") { - ssl = false; - rest = url.substr(5); - } else { - ssl = false; - rest = url; - } - - size_t path_pos = rest.find('/'); - std::string host_port; - if (path_pos != std::string::npos) { - host_port = rest.substr(0, path_pos); - path = rest.substr(path_pos); - } else { - host_port = rest; - path = "/"; - } - - size_t port_pos = host_port.find(':'); - if (port_pos != std::string::npos) { - host = host_port.substr(0, port_pos); - port = host_port.substr(port_pos + 1); - } else { - host = host_port; - port = ssl ? "443" : "80"; - } - } - - void doProxyConnect(const std::string& target_url) { - auto self = shared_from_this(); - bool ssl; - std::string host, port, path; - parseProxyUrl(target_url, ssl, host, port, path); - proxy_ssl_ = ssl; - proxy_host_ = host; - proxy_port_ = port; - proxy_path_ = path; - - auto exec = stream_.get_executor(); - proxy_resolver_ = std::make_unique(asio::make_strand(exec)); - - proxy_resolver_->async_resolve(host, port, - asio::bind_executor(exec, - [self](beast::error_code ec, tcp::resolver::results_type results) { - self->onProxyResolve(ec, results); - })); - } - - void onProxyResolve(beast::error_code ec, tcp::resolver::results_type results) { - if (ec) { - std::cerr << "[WS-Proxy] Resolve error: " << ec.message() << std::endl; - close(); - return; - } - - auto self = shared_from_this(); - auto exec = stream_.get_executor(); - - proxy_socket_ = std::make_unique(exec); - proxy_tcp_stream_ = std::make_unique(std::move(*proxy_socket_)); - proxy_socket_.reset(); - - proxy_tcp_stream_->expires_after(std::chrono::seconds(10)); - asio::async_connect(proxy_tcp_stream_->socket(), results, - asio::bind_executor(exec, - [self](beast::error_code ec, tcp::endpoint ep) { - self->onProxyConnect(ec, ep); - })); - } - - void onProxyConnect(beast::error_code ec, tcp::endpoint /*ep*/) { - if (ec) { - std::cerr << "[WS-Proxy] Connect error: " << ec.message() << std::endl; - close(); - return; - } - - auto self = shared_from_this(); - if (proxy_ssl_) { - proxy_ssl_ctx_ = std::make_unique(asio::ssl::context::tlsv12_client); - proxy_ssl_ctx_->set_verify_mode(asio::ssl::verify_none); - proxy_ssl_stream_ = std::make_unique>( - proxy_tcp_stream_->socket(), *proxy_ssl_ctx_); - - if (!SSL_set_tlsext_host_name(proxy_ssl_stream_->native_handle(), proxy_host_.c_str())) { - std::cerr << "[WS-Proxy] Failed to set SNI" << std::endl; - close(); - return; - } - - proxy_tcp_stream_->expires_after(std::chrono::seconds(10)); - auto exec = stream_.get_executor(); - proxy_ssl_stream_->async_handshake(asio::ssl::stream_base::client, - asio::bind_executor(exec, - [self](beast::error_code ec) { - self->onProxySslHandshake(ec); - })); - } else { - proxy_ws_stream_ = std::make_unique(proxy_tcp_stream_->socket()); - proxy_tcp_stream_->expires_never(); - - proxy_ws_stream_->set_option(websocket::stream_base::decorator( - [](websocket::request_type& req) { - req.set(http::field::user_agent, "UniServe-WsProxy"); - })); - - auto exec = self->stream_.get_executor(); - proxy_ws_stream_->async_handshake(proxy_host_, proxy_path_, - asio::bind_executor(exec, - [self](beast::error_code ec) { - self->onProxyHandshake(ec); - })); - } - } - - void onProxySslHandshake(beast::error_code ec) { - if (ec) { - std::cerr << "[WS-Proxy] SSL handshake error: " << ec.message() << std::endl; - close(); - return; - } - - auto self = shared_from_this(); - proxy_wss_stream_ = std::make_unique(*proxy_ssl_stream_); - proxy_tcp_stream_->expires_never(); - - proxy_wss_stream_->set_option(websocket::stream_base::decorator( - [](websocket::request_type& req) { - req.set(http::field::user_agent, "UniServe-WsProxy"); - })); - - auto exec = stream_.get_executor(); - proxy_wss_stream_->async_handshake(proxy_host_, proxy_path_, - asio::bind_executor(exec, - [self](beast::error_code ec) { - self->onProxyHandshake(ec); - })); - } - - void onProxyHandshake(beast::error_code ec) { - if (ec) { - std::cerr << "[WS-Proxy] Backend handshake error: " << ec.message() << std::endl; - close(); - return; - } - - proxy_connected_ = true; - proxy_closed_ = false; - std::cout << "[WS-Proxy] Connected to backend " << proxy_host_ << ":" << proxy_port_ - << proxy_path_ << std::endl; - - if (server_->ws_connect_handler_) { - server_->ws_connect_handler_(client_id_, getRemoteAddress()); - } - - // Start bidirectional bridging - doProxyReadClient(); - doProxyReadBackend(); - } - - // ---- Client → Backend direction ---- - - void doProxyReadClient() { - if (proxy_closed_) return; - auto self = shared_from_this(); - proxy_client_buffer_.consume(proxy_client_buffer_.size()); - - auto read_handler = [self](beast::error_code ec, std::size_t bytes_transferred) { - self->onProxyReadClient(ec, bytes_transferred); - }; - - if (ssl_enabled_ && wss_stream_) { - wss_stream_->async_read(proxy_client_buffer_, read_handler); - } else if (ws_stream_) { - ws_stream_->async_read(proxy_client_buffer_, read_handler); - } - } - - void onProxyReadClient(beast::error_code ec, std::size_t bytes_transferred) { - if (ec) { - if (ec != websocket::error::closed) { - std::cerr << "[WS-Proxy] Client read error: " << ec.message() << std::endl; - } - closeProxy(websocket::close_code::normal); - return; - } - - bool is_binary = false; - if (ssl_enabled_ && wss_stream_) { - is_binary = wss_stream_->got_binary(); - } else if (ws_stream_) { - is_binary = ws_stream_->got_binary(); - } - - std::string data = beast::buffers_to_string(proxy_client_buffer_.data()); - proxy_client_buffer_.consume(bytes_transferred); - - // Queue to backend - auto exec = stream_.get_executor(); - asio::post(exec, [self = shared_from_this(), data = std::move(data), is_binary]() { - self->proxy_to_backend_queue_.emplace_back(std::move(data), is_binary); - if (self->proxy_to_backend_queue_.size() == 1) { - self->doProxyWriteBackend(); - } - }); - - doProxyReadClient(); - } - - void doProxyWriteBackend() { - if (proxy_to_backend_queue_.empty() || !proxy_connected_ || proxy_closed_) return; - - auto self = shared_from_this(); - auto& [msg, is_binary] = proxy_to_backend_queue_.front(); - - auto write_handler = [self](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[WS-Proxy] Backend write error: " << ec.message() << std::endl; - self->closeProxy(websocket::close_code::internal_error); - return; - } - self->proxy_to_backend_queue_.pop_front(); - if (!self->proxy_to_backend_queue_.empty()) { - self->doProxyWriteBackend(); - } - }; - - if (proxy_ssl_ && proxy_wss_stream_) { - proxy_wss_stream_->binary(is_binary); - proxy_wss_stream_->async_write(asio::buffer(msg), write_handler); - } else if (proxy_ws_stream_) { - proxy_ws_stream_->binary(is_binary); - proxy_ws_stream_->async_write(asio::buffer(msg), write_handler); - } - } - - // ---- Backend → Client direction ---- - - void doProxyReadBackend() { - if (proxy_closed_) return; - auto self = shared_from_this(); - proxy_backend_buffer_.consume(proxy_backend_buffer_.size()); - - auto read_handler = [self](beast::error_code ec, std::size_t bytes_transferred) { - self->onProxyReadBackend(ec, bytes_transferred); - }; - - if (proxy_ssl_ && proxy_wss_stream_) { - proxy_wss_stream_->async_read(proxy_backend_buffer_, read_handler); - } else if (proxy_ws_stream_) { - proxy_ws_stream_->async_read(proxy_backend_buffer_, read_handler); - } - } - - void onProxyReadBackend(beast::error_code ec, std::size_t bytes_transferred) { - if (ec) { - if (ec != websocket::error::closed) { - std::cerr << "[WS-Proxy] Backend read error: " << ec.message() << std::endl; - } - closeProxy(websocket::close_code::normal); - return; - } - - bool is_binary = false; - if (proxy_ssl_ && proxy_wss_stream_) { - is_binary = proxy_wss_stream_->got_binary(); - } else if (proxy_ws_stream_) { - is_binary = proxy_ws_stream_->got_binary(); - } - - std::string data = beast::buffers_to_string(proxy_backend_buffer_.data()); - proxy_backend_buffer_.consume(bytes_transferred); - - // Queue to client - auto exec = stream_.get_executor(); - asio::post(exec, [self = shared_from_this(), data = std::move(data), is_binary]() { - self->proxy_to_client_queue_.emplace_back(std::move(data), is_binary); - if (self->proxy_to_client_queue_.size() == 1) { - self->doProxyWriteClient(); - } - }); - - doProxyReadBackend(); - } - - void doProxyWriteClient() { - if (proxy_to_client_queue_.empty() || !proxy_connected_ || proxy_closed_) return; - - auto self = shared_from_this(); - auto& [msg, is_binary] = proxy_to_client_queue_.front(); - - auto write_handler = [self](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[WS-Proxy] Client write error: " << ec.message() << std::endl; - self->closeProxy(websocket::close_code::internal_error); - return; - } - self->proxy_to_client_queue_.pop_front(); - if (!self->proxy_to_client_queue_.empty()) { - self->doProxyWriteClient(); - } - }; - - if (ssl_enabled_ && wss_stream_) { - wss_stream_->binary(is_binary); - wss_stream_->async_write(asio::buffer(msg), write_handler); - } else if (ws_stream_) { - ws_stream_->binary(is_binary); - ws_stream_->async_write(asio::buffer(msg), write_handler); - } - } - - void closeProxy(websocket::close_code code) { - if (proxy_closed_) return; - proxy_closed_ = true; - proxy_connected_ = false; - - beast::error_code ec; - // Close backend connection - if (proxy_ssl_ && proxy_wss_stream_) { - proxy_wss_stream_->close(code, ec); - } else if (proxy_ws_stream_) { - proxy_ws_stream_->close(code, ec); - } - if (proxy_ssl_stream_) { - proxy_ssl_stream_->shutdown(ec); - } - if (proxy_tcp_stream_) { - proxy_tcp_stream_->socket().shutdown(tcp::socket::shutdown_send, ec); - proxy_tcp_stream_->close(); - } - - proxy_to_backend_queue_.clear(); - proxy_to_client_queue_.clear(); - proxy_ws_stream_.reset(); - proxy_wss_stream_.reset(); - proxy_ssl_stream_.reset(); - proxy_ssl_ctx_.reset(); - proxy_tcp_stream_.reset(); - proxy_resolver_.reset(); - } - - // ======== Regular HTTP ======== - - void processRequest(http::request req) { - try { - auto resp = std::make_shared>( - server_->handleRequest(req)); - doWrite(resp); - } catch (const std::exception& e) { - std::cerr << "[Request] Handler exception: " << e.what() << std::endl; - auto err_resp = std::make_shared>(); - err_resp->version(req.version()); - err_resp->result(http::status::internal_server_error); - err_resp->set(http::field::content_type, "text/plain"); - err_resp->body() = "500 Internal Server Error\n"; - err_resp->prepare_payload(); - doWrite(err_resp); - } - } - - void doWrite(std::shared_ptr> response) { - auto self = shared_from_this(); - if (ssl_enabled_ && ssl_stream_) { - http::async_write(*ssl_stream_, *response, - [self, response](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[Write] Error to " << self->remote_ip_ << ":" << self->remote_port_ - << ": " << ec.message() << std::endl; - return; - } - if (!response->keep_alive()) self->close(); - else self->doRead(); - }); - } else { - http::async_write(stream_, *response, - [self, response](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[Write] Error to " << self->remote_ip_ << ":" << self->remote_port_ - << ": " << ec.message() << std::endl; - return; - } - if (!response->keep_alive()) self->close(); - else self->doRead(); - }); - } - } - - // ---- Client stream (inbound) ---- - tcp_stream stream_; - std::unique_ptr ssl_stream_; - std::unique_ptr ws_stream_; - std::unique_ptr wss_stream_; - - beast::flat_buffer buffer_; - beast::flat_buffer ws_buffer_; - http::request request_; - - HttpServer* server_; - bool ssl_enabled_; - - std::string remote_ip_ = "unknown"; - unsigned short remote_port_ = 0; - - bool is_websocket_; - bool is_ws_proxy_; - std::string client_id_; - std::string ws_path_; - WebSocketMessageHandler ws_handler_; - std::deque ws_send_queue_; - - // ---- Proxy state (outbound to backend) ---- - std::string ws_proxy_target_; - bool proxy_ssl_ = false; - bool proxy_connected_ = false; - bool proxy_closed_ = false; - std::string proxy_host_, proxy_port_, proxy_path_; - - std::unique_ptr proxy_resolver_; - std::unique_ptr proxy_socket_; - std::unique_ptr proxy_tcp_stream_; - std::unique_ptr proxy_ssl_ctx_; - std::unique_ptr> proxy_ssl_stream_; - std::unique_ptr proxy_ws_stream_; - std::unique_ptr proxy_wss_stream_; - - beast::flat_buffer proxy_client_buffer_; - beast::flat_buffer proxy_backend_buffer_; - std::deque> proxy_to_backend_queue_; - std::deque> proxy_to_client_queue_; -}; - -// ===================== Listener ===================== -class HttpServer::Listener : public std::enable_shared_from_this { -public: - Listener(asio::io_context& ioc, tcp::endpoint endpoint, HttpServer* server) - : ioc_(ioc), acceptor_(ioc), server_(server), ssl_enabled_(false) { - initAcceptor(endpoint); - } - - Listener(asio::io_context& ioc, tcp::endpoint endpoint, - asio::ssl::context& ssl_ctx, HttpServer* server) - : ioc_(ioc), acceptor_(ioc), ssl_ctx_(&ssl_ctx), server_(server), ssl_enabled_(true) { - initAcceptor(endpoint); - } - - void run() { - if (acceptor_.is_open()) doAccept(); - } - -private: - void initAcceptor(tcp::endpoint ep) { - beast::error_code ec; - acceptor_.open(ep.protocol(), ec); - if (!ec) { - acceptor_.set_option(asio::socket_base::reuse_address(true), ec); - acceptor_.bind(ep, ec); - if (!ec) acceptor_.listen(asio::socket_base::max_listen_connections, ec); - } - if (ec) throw HttpListenerException("Listener error: " + ec.message()); - } - - void doAccept() { - auto self = shared_from_this(); - acceptor_.async_accept( - asio::make_strand(ioc_), - [self](beast::error_code ec, tcp::socket socket) { - if (!ec) { - auto ep = socket.remote_endpoint(ec); - std::string ip = ec ? "unknown" : ep.address().to_string(); - unsigned short port = ec ? 0 : ep.port(); - std::cout << "[Listener] New connection from " << ip << ":" << port << std::endl; - - std::shared_ptr session; - if (self->ssl_enabled_ && self->ssl_ctx_) { - session = std::make_shared(std::move(socket), *self->ssl_ctx_, self->server_); - } else { - session = std::make_shared(std::move(socket), self->server_); - } - session->setRemoteInfo(ip, port); - self->server_->addSession(session); - session->run(); - } else { - std::cerr << "[Listener] Accept error: " << ec.message() << std::endl; - } - self->doAccept(); - }); - } - - asio::io_context& ioc_; - tcp::acceptor acceptor_; - asio::ssl::context* ssl_ctx_ = nullptr; - HttpServer* server_; - bool ssl_enabled_; -}; - -// ===================== HttpServer 实现 ===================== +// ===================== HttpServer 核心生命周期 ===================== HttpServer::HttpServer(const std::string& address, unsigned short port, unsigned int threads) : address_(address), port_(port), threads_(threads), ssl_enabled_(false), @@ -973,354 +122,34 @@ unsigned short HttpServer::getPort() const { return port_; } std::string HttpServer::getAddress() const { return address_; } bool HttpServer::isSSLEnabled() const { return ssl_enabled_; } -// ---- HTTP 路由注册 ---- -void HttpServer::get(const std::string& path, HttpRequestHandler handler) { - RouteEntry entry; - entry.path = path; - entry.handler = handler; - entry.is_wildcard = (path.find('*') != std::string::npos || - path.find('?') != std::string::npos); - if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); - get_handlers_.push_back(entry); +// ---- 会话管理 ---- +void HttpServer::addSession(const std::shared_ptr& session) { + std::lock_guard lock(sessions_mutex_); + sessions_.erase(std::remove_if(sessions_.begin(), sessions_.end(), + [](const std::weak_ptr& wp) { return wp.expired(); }), sessions_.end()); + sessions_.push_back(session); } -void HttpServer::post(const std::string& path, HttpRequestHandler handler) { - RouteEntry entry; - entry.path = path; - entry.handler = handler; - entry.is_wildcard = (path.find('*') != std::string::npos || - path.find('?') != std::string::npos); - if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); - post_handlers_.push_back(entry); +void HttpServer::removeSession(const Session* session) { + std::lock_guard lock(sessions_mutex_); + sessions_.erase(std::remove_if(sessions_.begin(), sessions_.end(), + [session](const std::weak_ptr& wp) { + auto sp = wp.lock(); + return !sp || sp.get() == session; + }), sessions_.end()); } -void HttpServer::put(const std::string& path, HttpRequestHandler handler) { - RouteEntry entry; - entry.path = path; - entry.handler = handler; - entry.is_wildcard = (path.find('*') != std::string::npos || - path.find('?') != std::string::npos); - if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); - put_handlers_.push_back(entry); +void HttpServer::closeAllSessions() { + std::lock_guard lock(sessions_mutex_); + sessions_.clear(); } -void HttpServer::del(const std::string& path, HttpRequestHandler handler) { - RouteEntry entry; - entry.path = path; - entry.handler = handler; - entry.is_wildcard = (path.find('*') != std::string::npos || - path.find('?') != std::string::npos); - if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); - delete_handlers_.push_back(entry); -} - -void HttpServer::options(const std::string& path, HttpRequestHandler handler) { - RouteEntry entry; - entry.path = path; - entry.handler = handler; - entry.is_wildcard = (path.find('*') != std::string::npos || - path.find('?') != std::string::npos); - if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); - options_handlers_.push_back(entry); -} - -void HttpServer::use(MiddlewareHandler m) { - middlewares_.push_back(m); -} - -void HttpServer::setStaticDirectory(const std::string& p) { - static_directory_ = p; -} - -// ---- WebSocket 路由注册 ---- -void HttpServer::ws(const std::string& path, WebSocketMessageHandler handler) { - WsRouteEntry entry; - entry.path = path; - entry.handler = handler; - ws_handlers_.push_back(entry); -} - -void HttpServer::ws_proxy(const std::string& path, WsProxyHandshakeHandler handshake) { - WsProxyRouteEntry entry; - entry.path = path; - entry.handshake = handshake; - ws_proxy_handlers_.push_back(entry); -} - -void HttpServer::onWsConnect(WebSocketConnectHandler handler) { - ws_connect_handler_ = handler; -} - -void HttpServer::onWsDisconnect(WebSocketDisconnectHandler handler) { - ws_disconnect_handler_ = handler; -} - -bool HttpServer::sendToWsClient(const std::string& clientId, const std::string& message) { - std::shared_ptr session; - { - std::lock_guard lock(ws_clients_mutex_); - auto it = ws_clients_.find(clientId); - if (it != ws_clients_.end()) { - session = it->second; - } +// ---- 调试工具 ---- +void HttpServer::debug::print_all_header_fields(const http::request& req) { + for (auto const& field : req) { + std::cout << "Name: " << field.name_string() + << ", Value: " << field.value() << std::endl; } - if (session) { - session->wsSend(message); - return true; - } - return false; -} - -size_t HttpServer::broadcastWs(const std::string& message) { - std::vector> clients; - { - std::lock_guard lock(ws_clients_mutex_); - for (auto& pair : ws_clients_) { - clients.push_back(pair.second); - } - } - for (auto& session : clients) { - session->wsSend(message); - } - return clients.size(); -} - -void HttpServer::disconnectWsClient(const std::string& clientId) { - std::lock_guard lock(ws_clients_mutex_); - ws_clients_.erase(clientId); -} - -size_t HttpServer::getWsClientCount() const { - std::lock_guard lock(ws_clients_mutex_); - return ws_clients_.size(); -} - -// ---- 通配符转换 ---- -std::regex HttpServer::wildcardToRegex(const std::string& pattern) { - std::string r; - r.reserve(pattern.size() * 2); - for (char c : pattern) { - if (c == '*') r += ".*"; - else if (c == '?') r += "."; - else if (c == '.' || c == '+' || c == '(' || c == ')' || c == '[' || c == ']' || - c == '{' || c == '}' || c == '^' || c == '$' || c == '|' || c == '\\') { - r += '\\'; - r += c; - } else r += c; - } - return std::regex(r, std::regex::icase); -} - -bool HttpServer::matchPath(const std::string& path, const std::regex& pattern) { - return std::regex_match(path, pattern); -} - -// ---- 查询参数解析 ---- -std::unordered_map HttpServer::parseQueryParams(const std::string& query) { - std::unordered_map params; - if (query.empty()) return params; - - std::vector pairs; - boost::split(pairs, query, boost::is_any_of("&")); - for (const auto& pair : pairs) { - std::vector kv; - boost::split(kv, pair, boost::is_any_of("=")); - if (kv.size() == 2) params[kv[0]] = kv[1]; - else if (kv.size() == 1) params[kv[0]] = ""; - } - return params; -} - -// ---- 路由查找 ---- -HttpRequestHandler HttpServer::findRoute(const std::string& method, const std::string& path) { - const std::vector* handlers = nullptr; - if (method == "GET") handlers = &get_handlers_; - else if (method == "POST") handlers = &post_handlers_; - else if (method == "PUT") handlers = &put_handlers_; - else if (method == "DELETE") handlers = &delete_handlers_; - else if (method == "OPTIONS") handlers = &options_handlers_; - else return nullptr; - - for (const auto& e : *handlers) { - if (!e.is_wildcard && e.path == path) return e.handler; - } - for (const auto& e : *handlers) { - if (e.is_wildcard && matchPath(path, e.pattern)) return e.handler; - } - return nullptr; -} - -bool HttpServer::routeExists(const std::string& method, const std::string& path) { - return findRoute(method, path) != nullptr; -} - -WebSocketMessageHandler HttpServer::findWsRoute(const std::string& path) { - for (const auto& e : ws_handlers_) { - if (e.path == path) return e.handler; - } - return nullptr; -} - -WsProxyHandshakeHandler HttpServer::findWsProxyRoute(const std::string& path) { - for (const auto& e : ws_proxy_handlers_) { - if (e.path == path) return e.handshake; - } - return nullptr; -} - -// ---- WebSocket 客户端管理 ---- -std::string HttpServer::generateWsClientId() { - return "ws_client_" + std::to_string(++ws_client_counter_); -} - -void HttpServer::addWsClient(const std::string& clientId, std::shared_ptr session) { - std::lock_guard lock(ws_clients_mutex_); - ws_clients_[clientId] = session; -} - -void HttpServer::removeWsClient(const std::string& clientId) { - std::lock_guard lock(ws_clients_mutex_); - ws_clients_.erase(clientId); -} - -std::shared_ptr HttpServer::getWsClient(const std::string& clientId) { - std::lock_guard lock(ws_clients_mutex_); - auto it = ws_clients_.find(clientId); - if (it != ws_clients_.end()) return it->second; - return nullptr; -} - -// ---- 请求处理核心 ---- -http::response HttpServer::handleRequest(const http::request& req) { - std::string target_path(req.target().data(), req.target().size()); - std::string query_string; - auto pos = target_path.find('?'); - if (pos != std::string::npos) { - query_string = target_path.substr(pos + 1); - target_path = target_path.substr(0, pos); - } - auto params = parseQueryParams(query_string); - - http::response res; - res.version(req.version()); - res.set(http::field::server, "Boost.Beast HttpServer"); - res.set(http::field::content_type, "text/plain"); - res.keep_alive(req.keep_alive()); - - bool middleware_handled = false; - for (auto& m : middlewares_) { - if (!m(req, res, params)) { - middleware_handled = true; - break; - } - } - if (middleware_handled) { - res.prepare_payload(); - return res; - } - - std::string method = http::to_string(req.method()); - - if (method == "OPTIONS") { - auto handler = findRoute("OPTIONS", target_path); - if (handler) { - try { - return handler(req, params); - } catch (const std::exception& e) { - std::cerr << "[Handler] Exception: " << e.what() << std::endl; - res.result(http::status::internal_server_error); - res.body() = "500 Internal Server Error\n"; - res.prepare_payload(); - return res; - } - } - - auto origin = req.find(http::field::origin); - if (origin != req.end()) { - std::vector all_methods = {"GET", "POST", "PUT", "DELETE", "OPTIONS"}; - std::vector allowed_methods; - - for (const auto& m : all_methods) { - if (m == "OPTIONS" || routeExists(m, target_path)) { - allowed_methods.push_back(m); - } - } - - if (!allowed_methods.empty() || findWsRoute(target_path)) { - if (allowed_methods.empty()) allowed_methods.push_back("GET"); - - std::string allow_methods; - for (size_t i = 0; i < allowed_methods.size(); ++i) { - if (i > 0) allow_methods += ", "; - allow_methods += allowed_methods[i]; - } - - res.result(http::status::ok); - res.set(http::field::allow, allow_methods); - res.set(http::field::access_control_allow_origin, - std::string(origin->value())); - res.set(http::field::access_control_allow_methods, allow_methods); - res.set(http::field::access_control_allow_headers, - "Content-Type, Authorization, X-Requested-With, Accept, Origin"); - res.set(http::field::access_control_max_age, "86400"); - res.body() = ""; - res.content_length(0); - res.prepare_payload(); - return res; - } - } - - res.result(http::status::not_found); - res.body() = "404 Not Found\n"; - res.prepare_payload(); - return res; - } - - auto handler = findRoute(method, target_path); - - if (!handler && !static_directory_.empty()) { - std::string file_path = static_directory_ + target_path; - std::ifstream file(file_path, std::ios::binary); - if (file.is_open()) { - std::stringstream buffer; - buffer << file.rdbuf(); - res.result(http::status::ok); - res.body() = buffer.str(); - - if (boost::algorithm::ends_with(file_path, ".html")) - res.set(http::field::content_type, "text/html"); - else if (boost::algorithm::ends_with(file_path, ".css")) - res.set(http::field::content_type, "text/css"); - else if (boost::algorithm::ends_with(file_path, ".js")) - res.set(http::field::content_type, "application/javascript"); - else if (boost::algorithm::ends_with(file_path, ".json")) - res.set(http::field::content_type, "application/json"); - else if (boost::algorithm::ends_with(file_path, ".png")) - res.set(http::field::content_type, "image/png"); - else if (boost::algorithm::ends_with(file_path, ".jpg") || - boost::algorithm::ends_with(file_path, ".jpeg")) - res.set(http::field::content_type, "image/jpeg"); - res.prepare_payload(); - return res; - } - } - - if (handler) { - try { - return handler(req, params); - } catch (const std::exception& e) { - std::cerr << "[Handler] Exception: " << e.what() << std::endl; - res.result(http::status::internal_server_error); - res.body() = "500 Internal Server Error\n"; - res.prepare_payload(); - return res; - } - } - - res.result(http::status::not_found); - res.body() = "404 Not Found\n"; - res.prepare_payload(); - return res; } void HttpServer::showStatus() const { @@ -1362,35 +191,5 @@ void HttpServer::showStatus() const { std::cout << "==========================" << std::endl; } -// ---- 会话管理 ---- -void HttpServer::addSession(const std::shared_ptr& session) { - std::lock_guard lock(sessions_mutex_); - sessions_.erase(std::remove_if(sessions_.begin(), sessions_.end(), - [](const std::weak_ptr& wp) { return wp.expired(); }), sessions_.end()); - sessions_.push_back(session); -} - -void HttpServer::removeSession(const Session* session) { - std::lock_guard lock(sessions_mutex_); - sessions_.erase(std::remove_if(sessions_.begin(), sessions_.end(), - [session](const std::weak_ptr& wp) { - auto sp = wp.lock(); - return !sp || sp.get() == session; - }), sessions_.end()); -} - -void HttpServer::closeAllSessions() { - std::lock_guard lock(sessions_mutex_); - sessions_.clear(); -} - -// ---- 调试工具 ---- -void HttpServer::debug::print_all_header_fields(const http::request& req) { - for (auto const& field : req) { - std::cout << "Name: " << field.name_string() - << ", Value: " << field.value() << std::endl; - } -} - } // namespace net } // namespace pmc diff --git a/libhttp/src/HttpServer_Listener.cpp b/libhttp/src/HttpServer_Listener.cpp new file mode 100644 index 0000000..6db6b0c --- /dev/null +++ b/libhttp/src/HttpServer_Listener.cpp @@ -0,0 +1,51 @@ +#include "us/detail/HttpServerImpl.hpp" + +namespace pmc { +namespace net { + +// ===================== Listener 实现 ===================== + +void HttpServer::Listener::run() { + if (acceptor_.is_open()) doAccept(); +} + +void HttpServer::Listener::initAcceptor(tcp::endpoint ep) { + beast::error_code ec; + acceptor_.open(ep.protocol(), ec); + if (!ec) { + acceptor_.set_option(asio::socket_base::reuse_address(true), ec); + acceptor_.bind(ep, ec); + if (!ec) acceptor_.listen(asio::socket_base::max_listen_connections, ec); + } + if (ec) throw HttpListenerException("Listener error: " + ec.message()); +} + +void HttpServer::Listener::doAccept() { + auto self = shared_from_this(); + acceptor_.async_accept( + asio::make_strand(ioc_), + [self](beast::error_code ec, tcp::socket socket) { + if (!ec) { + auto ep = socket.remote_endpoint(ec); + std::string ip = ec ? "unknown" : ep.address().to_string(); + unsigned short port = ec ? 0 : ep.port(); + std::cout << "[Listener] New connection from " << ip << ":" << port << std::endl; + + std::shared_ptr session; + if (self->ssl_enabled_ && self->ssl_ctx_) { + session = std::make_shared(std::move(socket), *self->ssl_ctx_, self->server_); + } else { + session = std::make_shared(std::move(socket), self->server_); + } + session->setRemoteInfo(ip, port); + self->server_->addSession(session); + session->run(); + } else { + std::cerr << "[Listener] Accept error: " << ec.message() << std::endl; + } + self->doAccept(); + }); +} + +} // namespace net +} // namespace pmc diff --git a/libhttp/src/HttpServer_Request.cpp b/libhttp/src/HttpServer_Request.cpp new file mode 100644 index 0000000..f5e5aa7 --- /dev/null +++ b/libhttp/src/HttpServer_Request.cpp @@ -0,0 +1,329 @@ +#include "us/detail/HttpServerImpl.hpp" + +namespace pmc { +namespace net { + +// ---- 请求处理核心 ---- +http::response HttpServer::handleRequest(const http::request& req) { + std::string target_path(req.target().data(), req.target().size()); + std::string query_string; + auto pos = target_path.find('?'); + if (pos != std::string::npos) { + query_string = target_path.substr(pos + 1); + target_path = target_path.substr(0, pos); + } + auto params = parseQueryParams(query_string); + + http::response res; + res.version(req.version()); + res.set(http::field::server, "Boost.Beast HttpServer"); + res.set(http::field::content_type, "text/plain"); + res.keep_alive(req.keep_alive()); + + bool middleware_handled = false; + for (auto& m : middlewares_) { + if (!m(req, res, params)) { + middleware_handled = true; + break; + } + } + if (middleware_handled) { + res.prepare_payload(); + return res; + } + + std::string method = http::to_string(req.method()); + + if (method == "OPTIONS") { + auto handler = findRoute("OPTIONS", target_path); + if (handler) { + try { + return handler(req, params); + } catch (const std::exception& e) { + std::cerr << "[Handler] Exception: " << e.what() << std::endl; + res.result(http::status::internal_server_error); + res.body() = "500 Internal Server Error\n"; + res.prepare_payload(); + return res; + } + } + + auto origin = req.find(http::field::origin); + if (origin != req.end()) { + std::vector all_methods = {"GET", "POST", "PUT", "DELETE", "OPTIONS"}; + std::vector allowed_methods; + + for (const auto& m : all_methods) { + if (m == "OPTIONS" || routeExists(m, target_path)) { + allowed_methods.push_back(m); + } + } + + if (!allowed_methods.empty() || findWsRoute(target_path)) { + if (allowed_methods.empty()) allowed_methods.push_back("GET"); + + std::string allow_methods; + for (size_t i = 0; i < allowed_methods.size(); ++i) { + if (i > 0) allow_methods += ", "; + allow_methods += allowed_methods[i]; + } + + res.result(http::status::ok); + res.set(http::field::allow, allow_methods); + res.set(http::field::access_control_allow_origin, + std::string(origin->value())); + res.set(http::field::access_control_allow_methods, allow_methods); + res.set(http::field::access_control_allow_headers, + "Content-Type, Authorization, X-Requested-With, Accept, Origin"); + res.set(http::field::access_control_max_age, "86400"); + res.body() = ""; + res.content_length(0); + res.prepare_payload(); + return res; + } + } + + res.result(http::status::not_found); + res.body() = "404 Not Found\n"; + res.prepare_payload(); + return res; + } + + auto handler = findRoute(method, target_path); + + if (!handler && !static_directory_.empty()) { + std::string file_path = static_directory_ + target_path; + std::ifstream file(file_path, std::ios::binary); + if (file.is_open()) { + std::stringstream buffer; + buffer << file.rdbuf(); + res.result(http::status::ok); + res.body() = buffer.str(); + + if (boost::algorithm::ends_with(file_path, ".html")) + res.set(http::field::content_type, "text/html"); + else if (boost::algorithm::ends_with(file_path, ".css")) + res.set(http::field::content_type, "text/css"); + else if (boost::algorithm::ends_with(file_path, ".js")) + res.set(http::field::content_type, "application/javascript"); + else if (boost::algorithm::ends_with(file_path, ".json")) + res.set(http::field::content_type, "application/json"); + else if (boost::algorithm::ends_with(file_path, ".png")) + res.set(http::field::content_type, "image/png"); + else if (boost::algorithm::ends_with(file_path, ".jpg") || + boost::algorithm::ends_with(file_path, ".jpeg")) + res.set(http::field::content_type, "image/jpeg"); + res.prepare_payload(); + return res; + } + } + + if (handler) { + try { + return handler(req, params); + } catch (const std::exception& e) { + std::cerr << "[Handler] Exception: " << e.what() << std::endl; + res.result(http::status::internal_server_error); + res.body() = "500 Internal Server Error\n"; + res.prepare_payload(); + return res; + } + } + + res.result(http::status::not_found); + res.body() = "404 Not Found\n"; + res.prepare_payload(); + return res; +} + +// ---- Session HTTP 处理 ---- +void HttpServer::Session::setRemoteInfo(const std::string& ip, unsigned short port) { + remote_ip_ = ip; + remote_port_ = port; +} + +void HttpServer::Session::doSSLHandshake() { + auto self = shared_from_this(); + std::cout << "[SSL] Starting handshake for " + << remote_ip_ << ":" << remote_port_ << std::endl; + + ssl_stream_->async_handshake( + asio::ssl::stream_base::server, + [self](beast::error_code ec) { + if (ec) { + std::cerr << "[SSL] Handshake error: " + << self->remote_ip_ << ":" << self->remote_port_ + << " - " << ec.message() + << " (code=" << ec.value() << ")" << std::endl; + ERR_print_errors_fp(stderr); + return; + } + std::cout << "[SSL] Handshake succeeded: " + << self->remote_ip_ << ":" << self->remote_port_ << std::endl; + self->doRead(); + }); +} + +void HttpServer::Session::doRead() { + auto self = shared_from_this(); + request_ = {}; + buffer_.consume(buffer_.size()); + + if (ssl_enabled_ && ssl_stream_) { + http::async_read(*ssl_stream_, buffer_, request_, + [self](beast::error_code ec, std::size_t) { + self->handleRead(ec); + }); + } else { + http::async_read(stream_, buffer_, request_, + [self](beast::error_code ec, std::size_t) { + self->handleRead(ec); + }); + } +} + +bool HttpServer::Session::isWebSocketUpgrade(const http::request& req) { + if (req.method() != http::verb::get) return false; + + auto upgrade = req.find(http::field::upgrade); + if (upgrade == req.end()) return false; + + std::string upgrade_val(upgrade->value()); + boost::to_lower(upgrade_val); + boost::trim(upgrade_val); + if (upgrade_val != "websocket") return false; + + auto connection = req.find(http::field::connection); + if (connection == req.end()) return false; + + std::string conn_val(connection->value()); + boost::to_lower(conn_val); + return conn_val.find("upgrade") != std::string::npos; +} + +void HttpServer::Session::handleRead(beast::error_code ec) { + if (!ec) { + if (isWebSocketUpgrade(request_)) { + std::string path = std::string(request_.target()); + auto qpos = path.find('?'); + if (qpos != std::string::npos) path = path.substr(0, qpos); + + // Check proxy routes first (higher priority) + auto proxy_handshake = server_->findWsProxyRoute(path); + if (proxy_handshake) { + std::string target = proxy_handshake(request_); + if (target.empty()) { + // Rejected by handshake callback → 403 + auto resp = std::make_shared>(); + resp->version(request_.version()); + resp->result(http::status::forbidden); + resp->set(http::field::content_type, "text/plain"); + resp->set(http::field::connection, "close"); + resp->body() = "403 Forbidden\n"; + resp->prepare_payload(); + doWrite(resp); + return; + } + ws_path_ = path; + is_ws_proxy_ = true; + ws_proxy_target_ = target; + if (ssl_enabled_) { + doWssUpgrade(); + } else { + doWsUpgrade(); + } + return; + } + + auto ws_handler = server_->findWsRoute(path); + if (ws_handler) { + ws_handler_ = ws_handler; + ws_path_ = path; + if (ssl_enabled_) { + doWssUpgrade(); + } else { + doWsUpgrade(); + } + return; + } + } + + auto req_copy = std::move(request_); + processRequest(std::move(req_copy)); + } else if (ec != http::error::end_of_stream) { + std::cerr << "[Read] Error from " << remote_ip_ << ":" << remote_port_ + << ": " << ec.message() << std::endl; + } +} + +void HttpServer::Session::processRequest(http::request req) { + try { + auto resp = std::make_shared>( + server_->handleRequest(req)); + doWrite(resp); + } catch (const std::exception& e) { + std::cerr << "[Request] Handler exception: " << e.what() << std::endl; + auto err_resp = std::make_shared>(); + err_resp->version(req.version()); + err_resp->result(http::status::internal_server_error); + err_resp->set(http::field::content_type, "text/plain"); + err_resp->body() = "500 Internal Server Error\n"; + err_resp->prepare_payload(); + doWrite(err_resp); + } +} + +void HttpServer::Session::doWrite(std::shared_ptr> response) { + auto self = shared_from_this(); + if (ssl_enabled_ && ssl_stream_) { + http::async_write(*ssl_stream_, *response, + [self, response](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[Write] Error to " << self->remote_ip_ << ":" << self->remote_port_ + << ": " << ec.message() << std::endl; + return; + } + if (!response->keep_alive()) self->close(); + else self->doRead(); + }); + } else { + http::async_write(stream_, *response, + [self, response](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[Write] Error to " << self->remote_ip_ << ":" << self->remote_port_ + << ": " << ec.message() << std::endl; + return; + } + if (!response->keep_alive()) self->close(); + else self->doRead(); + }); + } +} + +void HttpServer::Session::close() { + beast::error_code ec; + closeProxy(websocket::close_code::going_away); + if (is_websocket_) { + if (ssl_enabled_ && wss_stream_) { + wss_stream_->close(websocket::close_code::normal, ec); + } else if (ws_stream_) { + ws_stream_->close(websocket::close_code::normal, ec); + } + } + if (ssl_stream_) { + ssl_stream_->shutdown(ec); + } + stream_.socket().shutdown(tcp::socket::shutdown_send, ec); + stream_.close(); +} + +void HttpServer::Session::run() { + if (ssl_enabled_) { + doSSLHandshake(); + } else { + doRead(); + } +} + +} // namespace net +} // namespace pmc diff --git a/libhttp/src/HttpServer_Routes.cpp b/libhttp/src/HttpServer_Routes.cpp new file mode 100644 index 0000000..6683a46 --- /dev/null +++ b/libhttp/src/HttpServer_Routes.cpp @@ -0,0 +1,125 @@ +#include "us/detail/HttpServerImpl.hpp" + +namespace pmc { +namespace net { + +// ---- HTTP 路由注册 ---- +void HttpServer::get(const std::string& path, HttpRequestHandler handler) { + RouteEntry entry; + entry.path = path; + entry.handler = handler; + entry.is_wildcard = (path.find('*') != std::string::npos || + path.find('?') != std::string::npos); + if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); + get_handlers_.push_back(entry); +} + +void HttpServer::post(const std::string& path, HttpRequestHandler handler) { + RouteEntry entry; + entry.path = path; + entry.handler = handler; + entry.is_wildcard = (path.find('*') != std::string::npos || + path.find('?') != std::string::npos); + if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); + post_handlers_.push_back(entry); +} + +void HttpServer::put(const std::string& path, HttpRequestHandler handler) { + RouteEntry entry; + entry.path = path; + entry.handler = handler; + entry.is_wildcard = (path.find('*') != std::string::npos || + path.find('?') != std::string::npos); + if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); + put_handlers_.push_back(entry); +} + +void HttpServer::del(const std::string& path, HttpRequestHandler handler) { + RouteEntry entry; + entry.path = path; + entry.handler = handler; + entry.is_wildcard = (path.find('*') != std::string::npos || + path.find('?') != std::string::npos); + if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); + delete_handlers_.push_back(entry); +} + +void HttpServer::options(const std::string& path, HttpRequestHandler handler) { + RouteEntry entry; + entry.path = path; + entry.handler = handler; + entry.is_wildcard = (path.find('*') != std::string::npos || + path.find('?') != std::string::npos); + if (entry.is_wildcard) entry.pattern = wildcardToRegex(path); + options_handlers_.push_back(entry); +} + +void HttpServer::use(MiddlewareHandler m) { + middlewares_.push_back(m); +} + +void HttpServer::setStaticDirectory(const std::string& p) { + static_directory_ = p; +} + +// ---- 通配符转换 ---- +std::regex HttpServer::wildcardToRegex(const std::string& pattern) { + std::string r; + r.reserve(pattern.size() * 2); + for (char c : pattern) { + if (c == '*') r += ".*"; + else if (c == '?') r += "."; + else if (c == '.' || c == '+' || c == '(' || c == ')' || c == '[' || c == ']' || + c == '{' || c == '}' || c == '^' || c == '$' || c == '|' || c == '\\') { + r += '\\'; + r += c; + } else r += c; + } + return std::regex(r, std::regex::icase); +} + +bool HttpServer::matchPath(const std::string& path, const std::regex& pattern) { + return std::regex_match(path, pattern); +} + +// ---- 查询参数解析 ---- +std::unordered_map HttpServer::parseQueryParams(const std::string& query) { + std::unordered_map params; + if (query.empty()) return params; + + std::vector pairs; + boost::split(pairs, query, boost::is_any_of("&")); + for (const auto& pair : pairs) { + std::vector kv; + boost::split(kv, pair, boost::is_any_of("=")); + if (kv.size() == 2) params[kv[0]] = kv[1]; + else if (kv.size() == 1) params[kv[0]] = ""; + } + return params; +} + +// ---- 路由查找 ---- +HttpRequestHandler HttpServer::findRoute(const std::string& method, const std::string& path) { + const std::vector* handlers = nullptr; + if (method == "GET") handlers = &get_handlers_; + else if (method == "POST") handlers = &post_handlers_; + else if (method == "PUT") handlers = &put_handlers_; + else if (method == "DELETE") handlers = &delete_handlers_; + else if (method == "OPTIONS") handlers = &options_handlers_; + else return nullptr; + + for (const auto& e : *handlers) { + if (!e.is_wildcard && e.path == path) return e.handler; + } + for (const auto& e : *handlers) { + if (e.is_wildcard && matchPath(path, e.pattern)) return e.handler; + } + return nullptr; +} + +bool HttpServer::routeExists(const std::string& method, const std::string& path) { + return findRoute(method, path) != nullptr; +} + +} // namespace net +} // namespace pmc diff --git a/libhttp/src/HttpServer_Ws.cpp b/libhttp/src/HttpServer_Ws.cpp new file mode 100644 index 0000000..d037a97 --- /dev/null +++ b/libhttp/src/HttpServer_Ws.cpp @@ -0,0 +1,106 @@ +#include "us/detail/HttpServerImpl.hpp" + +namespace pmc { +namespace net { + +// ---- WebSocket 路由注册 ---- +void HttpServer::ws(const std::string& path, WebSocketMessageHandler handler) { + WsRouteEntry entry; + entry.path = path; + entry.handler = handler; + ws_handlers_.push_back(entry); +} + +void HttpServer::ws_proxy(const std::string& path, WsProxyHandshakeHandler handshake) { + WsProxyRouteEntry entry; + entry.path = path; + entry.handshake = handshake; + ws_proxy_handlers_.push_back(entry); +} + +void HttpServer::onWsConnect(WebSocketConnectHandler handler) { + ws_connect_handler_ = handler; +} + +void HttpServer::onWsDisconnect(WebSocketDisconnectHandler handler) { + ws_disconnect_handler_ = handler; +} + +WebSocketMessageHandler HttpServer::findWsRoute(const std::string& path) { + for (const auto& e : ws_handlers_) { + if (e.path == path) return e.handler; + } + return nullptr; +} + +WsProxyHandshakeHandler HttpServer::findWsProxyRoute(const std::string& path) { + for (const auto& e : ws_proxy_handlers_) { + if (e.path == path) return e.handshake; + } + return nullptr; +} + +// ---- WebSocket 客户端管理 ---- +std::string HttpServer::generateWsClientId() { + return "ws_client_" + std::to_string(++ws_client_counter_); +} + +void HttpServer::addWsClient(const std::string& clientId, std::shared_ptr session) { + std::lock_guard lock(ws_clients_mutex_); + ws_clients_[clientId] = session; +} + +void HttpServer::removeWsClient(const std::string& clientId) { + std::lock_guard lock(ws_clients_mutex_); + ws_clients_.erase(clientId); +} + +std::shared_ptr HttpServer::getWsClient(const std::string& clientId) { + std::lock_guard lock(ws_clients_mutex_); + auto it = ws_clients_.find(clientId); + if (it != ws_clients_.end()) return it->second; + return nullptr; +} + +bool HttpServer::sendToWsClient(const std::string& clientId, const std::string& message) { + std::shared_ptr session; + { + std::lock_guard lock(ws_clients_mutex_); + auto it = ws_clients_.find(clientId); + if (it != ws_clients_.end()) { + session = it->second; + } + } + if (session) { + session->wsSend(message); + return true; + } + return false; +} + +size_t HttpServer::broadcastWs(const std::string& message) { + std::vector> clients; + { + std::lock_guard lock(ws_clients_mutex_); + for (auto& pair : ws_clients_) { + clients.push_back(pair.second); + } + } + for (auto& session : clients) { + session->wsSend(message); + } + return clients.size(); +} + +void HttpServer::disconnectWsClient(const std::string& clientId) { + std::lock_guard lock(ws_clients_mutex_); + ws_clients_.erase(clientId); +} + +size_t HttpServer::getWsClientCount() const { + std::lock_guard lock(ws_clients_mutex_); + return ws_clients_.size(); +} + +} // namespace net +} // namespace pmc diff --git a/libhttp/src/HttpServer_WsSession.cpp b/libhttp/src/HttpServer_WsSession.cpp new file mode 100644 index 0000000..25eb7c9 --- /dev/null +++ b/libhttp/src/HttpServer_WsSession.cpp @@ -0,0 +1,506 @@ +#include "us/detail/HttpServerImpl.hpp" + +namespace pmc { +namespace net { + +// ===================== Session WebSocket 处理 ===================== + +void HttpServer::Session::wsSend(const std::string& message) { + auto exec = stream_.get_executor(); + asio::post(exec, [self = shared_from_this(), message]() { + self->ws_send_queue_.push_back(message); + if (self->ws_send_queue_.size() == 1) { + self->doWsWrite(); + } + }); +} + +void HttpServer::Session::doWsUpgrade() { + auto self = shared_from_this(); + ws_stream_ = std::make_unique(stream_); + ws_stream_->text(true); + + ws_stream_->async_accept( + request_, + [self](beast::error_code ec) { + self->onWsAccept(ec); + }); +} + +void HttpServer::Session::doWssUpgrade() { + auto self = shared_from_this(); + wss_stream_ = std::make_unique(*ssl_stream_); + wss_stream_->text(true); + + wss_stream_->async_accept( + request_, + [self](beast::error_code ec) { + self->onWsAccept(ec); + }); +} + +void HttpServer::Session::onWsAccept(beast::error_code ec) { + if (ec) { + std::cerr << "[WS] Accept error from " << remote_ip_ << ":" << remote_port_ + << ": " << ec.message() << std::endl; + return; + } + + is_websocket_ = true; + client_id_ = server_->generateWsClientId(); + server_->addWsClient(client_id_, shared_from_this()); + + std::cout << "[WS] Client " << client_id_ << " connected from " + << remote_ip_ << ":" << remote_port_ << " at " << ws_path_ << std::endl; + + if (is_ws_proxy_) { + std::cout << "[WS-Proxy] Proxying " << ws_path_ << " -> " << ws_proxy_target_ << std::endl; + doProxyConnect(ws_proxy_target_); + return; + } + + if (server_->ws_connect_handler_) { + server_->ws_connect_handler_(client_id_, getRemoteAddress()); + } + + doWsRead(); +} + +void HttpServer::Session::doWsRead() { + auto self = shared_from_this(); + ws_buffer_.consume(ws_buffer_.size()); + + if (ssl_enabled_ && wss_stream_) { + wss_stream_->async_read( + ws_buffer_, + [self](beast::error_code ec, std::size_t bytes_transferred) { + self->onWsRead(ec, bytes_transferred); + }); + } else if (ws_stream_) { + ws_stream_->async_read( + ws_buffer_, + [self](beast::error_code ec, std::size_t bytes_transferred) { + self->onWsRead(ec, bytes_transferred); + }); + } +} + +void HttpServer::Session::onWsRead(beast::error_code ec, std::size_t bytes_transferred) { + if (ec == websocket::error::closed) { + onWsClose(); + return; + } + if (ec) { + std::cerr << "[WS] Read error from " << client_id_ << ": " << ec.message() << std::endl; + onWsClose(); + return; + } + + std::string message = beast::buffers_to_string(ws_buffer_.data()); + ws_buffer_.consume(bytes_transferred); + + if (ws_handler_) { + auto self = shared_from_this(); + ws_handler_(client_id_, message, + [self](const std::string& response) { + self->wsSend(response); + }); + } + + doWsRead(); +} + +void HttpServer::Session::doWsWrite() { + if (ws_send_queue_.empty()) return; + + auto self = shared_from_this(); + std::string& message = ws_send_queue_.front(); + + auto write_handler = [self](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[WS] Write error to " << self->client_id_ << ": " << ec.message() << std::endl; + self->onWsClose(); + return; + } + self->ws_send_queue_.pop_front(); + if (!self->ws_send_queue_.empty()) { + self->doWsWrite(); + } + }; + + if (ssl_enabled_ && wss_stream_) { + wss_stream_->text(true); + wss_stream_->async_write(asio::buffer(message), write_handler); + } else if (ws_stream_) { + ws_stream_->text(true); + ws_stream_->async_write(asio::buffer(message), write_handler); + } +} + +void HttpServer::Session::onWsClose() { + if (client_id_.empty()) return; + std::cout << "[WS] Client " << client_id_ << " disconnected" << std::endl; + + closeProxy(websocket::close_code::normal); + server_->removeWsClient(client_id_); + + if (server_->ws_disconnect_handler_) { + server_->ws_disconnect_handler_(client_id_); + } + client_id_.clear(); + is_websocket_ = false; + is_ws_proxy_ = false; +} + +// ===================== WebSocket Proxy (双向桥接) ===================== + +void HttpServer::Session::parseProxyUrl(const std::string& url, bool& ssl, std::string& host, + std::string& port, std::string& path) { + std::string rest; + if (url.size() >= 6 && url.substr(0, 6) == "wss://") { + ssl = true; + rest = url.substr(6); + } else if (url.size() >= 5 && url.substr(0, 5) == "ws://") { + ssl = false; + rest = url.substr(5); + } else { + ssl = false; + rest = url; + } + + size_t path_pos = rest.find('/'); + std::string host_port; + if (path_pos != std::string::npos) { + host_port = rest.substr(0, path_pos); + path = rest.substr(path_pos); + } else { + host_port = rest; + path = "/"; + } + + size_t port_pos = host_port.find(':'); + if (port_pos != std::string::npos) { + host = host_port.substr(0, port_pos); + port = host_port.substr(port_pos + 1); + } else { + host = host_port; + port = ssl ? "443" : "80"; + } +} + +void HttpServer::Session::doProxyConnect(const std::string& target_url) { + auto self = shared_from_this(); + bool ssl; + std::string host, port, path; + parseProxyUrl(target_url, ssl, host, port, path); + proxy_ssl_ = ssl; + proxy_host_ = host; + proxy_port_ = port; + proxy_path_ = path; + + auto exec = stream_.get_executor(); + proxy_resolver_ = std::make_unique(asio::make_strand(exec)); + + proxy_resolver_->async_resolve(host, port, + asio::bind_executor(exec, + [self](beast::error_code ec, tcp::resolver::results_type results) { + self->onProxyResolve(ec, results); + })); +} + +void HttpServer::Session::onProxyResolve(beast::error_code ec, tcp::resolver::results_type results) { + if (ec) { + std::cerr << "[WS-Proxy] Resolve error: " << ec.message() << std::endl; + close(); + return; + } + + auto self = shared_from_this(); + auto exec = stream_.get_executor(); + + proxy_socket_ = std::make_unique(exec); + proxy_tcp_stream_ = std::make_unique(std::move(*proxy_socket_)); + proxy_socket_.reset(); + + proxy_tcp_stream_->expires_after(std::chrono::seconds(10)); + asio::async_connect(proxy_tcp_stream_->socket(), results, + asio::bind_executor(exec, + [self](beast::error_code ec, tcp::endpoint ep) { + self->onProxyConnect(ec, ep); + })); +} + +void HttpServer::Session::onProxyConnect(beast::error_code ec, tcp::endpoint /*ep*/) { + if (ec) { + std::cerr << "[WS-Proxy] Connect error: " << ec.message() << std::endl; + close(); + return; + } + + auto self = shared_from_this(); + if (proxy_ssl_) { + proxy_ssl_ctx_ = std::make_unique(asio::ssl::context::tlsv12_client); + proxy_ssl_ctx_->set_verify_mode(asio::ssl::verify_none); + proxy_ssl_stream_ = std::make_unique>( + proxy_tcp_stream_->socket(), *proxy_ssl_ctx_); + + if (!SSL_set_tlsext_host_name(proxy_ssl_stream_->native_handle(), proxy_host_.c_str())) { + std::cerr << "[WS-Proxy] Failed to set SNI" << std::endl; + close(); + return; + } + + proxy_tcp_stream_->expires_after(std::chrono::seconds(10)); + auto exec = stream_.get_executor(); + proxy_ssl_stream_->async_handshake(asio::ssl::stream_base::client, + asio::bind_executor(exec, + [self](beast::error_code ec) { + self->onProxySslHandshake(ec); + })); + } else { + proxy_ws_stream_ = std::make_unique(proxy_tcp_stream_->socket()); + proxy_tcp_stream_->expires_never(); + + proxy_ws_stream_->set_option(websocket::stream_base::decorator( + [](websocket::request_type& req) { + req.set(http::field::user_agent, "UniServe-WsProxy"); + })); + + auto exec = self->stream_.get_executor(); + proxy_ws_stream_->async_handshake(proxy_host_, proxy_path_, + asio::bind_executor(exec, + [self](beast::error_code ec) { + self->onProxyHandshake(ec); + })); + } +} + +void HttpServer::Session::onProxySslHandshake(beast::error_code ec) { + if (ec) { + std::cerr << "[WS-Proxy] SSL handshake error: " << ec.message() << std::endl; + close(); + return; + } + + auto self = shared_from_this(); + proxy_wss_stream_ = std::make_unique(*proxy_ssl_stream_); + proxy_tcp_stream_->expires_never(); + + proxy_wss_stream_->set_option(websocket::stream_base::decorator( + [](websocket::request_type& req) { + req.set(http::field::user_agent, "UniServe-WsProxy"); + })); + + auto exec = stream_.get_executor(); + proxy_wss_stream_->async_handshake(proxy_host_, proxy_path_, + asio::bind_executor(exec, + [self](beast::error_code ec) { + self->onProxyHandshake(ec); + })); +} + +void HttpServer::Session::onProxyHandshake(beast::error_code ec) { + if (ec) { + std::cerr << "[WS-Proxy] Backend handshake error: " << ec.message() << std::endl; + close(); + return; + } + + proxy_connected_ = true; + proxy_closed_ = false; + std::cout << "[WS-Proxy] Connected to backend " << proxy_host_ << ":" << proxy_port_ + << proxy_path_ << std::endl; + + if (server_->ws_connect_handler_) { + server_->ws_connect_handler_(client_id_, getRemoteAddress()); + } + + // Start bidirectional bridging + doProxyReadClient(); + doProxyReadBackend(); +} + +// ---- Client → Backend direction ---- + +void HttpServer::Session::doProxyReadClient() { + if (proxy_closed_) return; + auto self = shared_from_this(); + proxy_client_buffer_.consume(proxy_client_buffer_.size()); + + auto read_handler = [self](beast::error_code ec, std::size_t bytes_transferred) { + self->onProxyReadClient(ec, bytes_transferred); + }; + + if (ssl_enabled_ && wss_stream_) { + wss_stream_->async_read(proxy_client_buffer_, read_handler); + } else if (ws_stream_) { + ws_stream_->async_read(proxy_client_buffer_, read_handler); + } +} + +void HttpServer::Session::onProxyReadClient(beast::error_code ec, std::size_t bytes_transferred) { + if (ec) { + if (ec != websocket::error::closed) { + std::cerr << "[WS-Proxy] Client read error: " << ec.message() << std::endl; + } + closeProxy(websocket::close_code::normal); + return; + } + + bool is_binary = false; + if (ssl_enabled_ && wss_stream_) { + is_binary = wss_stream_->got_binary(); + } else if (ws_stream_) { + is_binary = ws_stream_->got_binary(); + } + + std::string data = beast::buffers_to_string(proxy_client_buffer_.data()); + proxy_client_buffer_.consume(bytes_transferred); + + // Queue to backend + auto exec = stream_.get_executor(); + asio::post(exec, [self = shared_from_this(), data = std::move(data), is_binary]() { + self->proxy_to_backend_queue_.emplace_back(std::move(data), is_binary); + if (self->proxy_to_backend_queue_.size() == 1) { + self->doProxyWriteBackend(); + } + }); + + doProxyReadClient(); +} + +void HttpServer::Session::doProxyWriteBackend() { + if (proxy_to_backend_queue_.empty() || !proxy_connected_ || proxy_closed_) return; + + auto self = shared_from_this(); + auto& [msg, is_binary] = proxy_to_backend_queue_.front(); + + auto write_handler = [self](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[WS-Proxy] Backend write error: " << ec.message() << std::endl; + self->closeProxy(websocket::close_code::internal_error); + return; + } + self->proxy_to_backend_queue_.pop_front(); + if (!self->proxy_to_backend_queue_.empty()) { + self->doProxyWriteBackend(); + } + }; + + if (proxy_ssl_ && proxy_wss_stream_) { + proxy_wss_stream_->binary(is_binary); + proxy_wss_stream_->async_write(asio::buffer(msg), write_handler); + } else if (proxy_ws_stream_) { + proxy_ws_stream_->binary(is_binary); + proxy_ws_stream_->async_write(asio::buffer(msg), write_handler); + } +} + +// ---- Backend → Client direction ---- + +void HttpServer::Session::doProxyReadBackend() { + if (proxy_closed_) return; + auto self = shared_from_this(); + proxy_backend_buffer_.consume(proxy_backend_buffer_.size()); + + auto read_handler = [self](beast::error_code ec, std::size_t bytes_transferred) { + self->onProxyReadBackend(ec, bytes_transferred); + }; + + if (proxy_ssl_ && proxy_wss_stream_) { + proxy_wss_stream_->async_read(proxy_backend_buffer_, read_handler); + } else if (proxy_ws_stream_) { + proxy_ws_stream_->async_read(proxy_backend_buffer_, read_handler); + } +} + +void HttpServer::Session::onProxyReadBackend(beast::error_code ec, std::size_t bytes_transferred) { + if (ec) { + if (ec != websocket::error::closed) { + std::cerr << "[WS-Proxy] Backend read error: " << ec.message() << std::endl; + } + closeProxy(websocket::close_code::normal); + return; + } + + bool is_binary = false; + if (proxy_ssl_ && proxy_wss_stream_) { + is_binary = proxy_wss_stream_->got_binary(); + } else if (proxy_ws_stream_) { + is_binary = proxy_ws_stream_->got_binary(); + } + + std::string data = beast::buffers_to_string(proxy_backend_buffer_.data()); + proxy_backend_buffer_.consume(bytes_transferred); + + // Queue to client + auto exec = stream_.get_executor(); + asio::post(exec, [self = shared_from_this(), data = std::move(data), is_binary]() { + self->proxy_to_client_queue_.emplace_back(std::move(data), is_binary); + if (self->proxy_to_client_queue_.size() == 1) { + self->doProxyWriteClient(); + } + }); + + doProxyReadBackend(); +} + +void HttpServer::Session::doProxyWriteClient() { + if (proxy_to_client_queue_.empty() || !proxy_connected_ || proxy_closed_) return; + + auto self = shared_from_this(); + auto& [msg, is_binary] = proxy_to_client_queue_.front(); + + auto write_handler = [self](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[WS-Proxy] Client write error: " << ec.message() << std::endl; + self->closeProxy(websocket::close_code::internal_error); + return; + } + self->proxy_to_client_queue_.pop_front(); + if (!self->proxy_to_client_queue_.empty()) { + self->doProxyWriteClient(); + } + }; + + if (ssl_enabled_ && wss_stream_) { + wss_stream_->binary(is_binary); + wss_stream_->async_write(asio::buffer(msg), write_handler); + } else if (ws_stream_) { + ws_stream_->binary(is_binary); + ws_stream_->async_write(asio::buffer(msg), write_handler); + } +} + +void HttpServer::Session::closeProxy(websocket::close_code code) { + if (proxy_closed_) return; + proxy_closed_ = true; + proxy_connected_ = false; + + beast::error_code ec; + // Close backend connection + if (proxy_ssl_ && proxy_wss_stream_) { + proxy_wss_stream_->close(code, ec); + } else if (proxy_ws_stream_) { + proxy_ws_stream_->close(code, ec); + } + if (proxy_ssl_stream_) { + proxy_ssl_stream_->shutdown(ec); + } + if (proxy_tcp_stream_) { + proxy_tcp_stream_->socket().shutdown(tcp::socket::shutdown_send, ec); + proxy_tcp_stream_->close(); + } + + proxy_to_backend_queue_.clear(); + proxy_to_client_queue_.clear(); + proxy_ws_stream_.reset(); + proxy_wss_stream_.reset(); + proxy_ssl_stream_.reset(); + proxy_ssl_ctx_.reset(); + proxy_tcp_stream_.reset(); + proxy_resolver_.reset(); +} + +} // namespace net +} // namespace pmc