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