From ba9a77fe66c884f7c786b03a6246eafdfd274ad9 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 1 Aug 2026 04:43:33 +0800 Subject: [PATCH] feat(libhttp): integrate WebSocket support into HttpServer Merge WebSocketServer functionality into HttpServer so a single port can serve both HTTP(S) and WebSocket(S) endpoints while preserving full backward compatibility for existing HTTP interfaces. Key changes: - Add WebSocket route registration: ws(), onWsConnect(), onWsDisconnect() - Add client management: sendToWsClient(), broadcastWs(), disconnectWsClient() - Rewrite Session class using beast::tcp_stream for unified HTTP/WS handling - Support automatic ws:// and wss:// upgrades on HTTP/HTTPS ports - Thread-safe message queue with asio::post for cross-thread sends - Update Makefile to link OpenSSL and pthread for TLS/WebSocket support - Add test_server.cpp demonstrating HTTP + WebSocket mixed usage --- libhttp/Makefile | 4 +- libhttp/inc/us/HttpServer.hpp | 53 +++- libhttp/src/HttpServer.cpp | 486 +++++++++++++++++++++++++++------- libhttp/test_server.cpp | 145 ++++++++++ 4 files changed, 580 insertions(+), 108 deletions(-) create mode 100644 libhttp/test_server.cpp diff --git a/libhttp/Makefile b/libhttp/Makefile index 5fc8f46..7e744b0 100644 --- a/libhttp/Makefile +++ b/libhttp/Makefile @@ -5,9 +5,9 @@ INSTL_TDIR := /lib INCLU_TDIR := /usr/include INCLU_SDIR := inc CPP := g++ -INCS := -I inc +INCS := -I inc -I/opt/homebrew/include CPPFLAGS := --std=c++20 -O3 -pipe -Wall -Werror -LIBS := +LIBS := -L/opt/homebrew/lib -lssl -lcrypto -lpthread BUILD_TDIR := build $(shell mkdir -p $(BUILD_TDIR)) diff --git a/libhttp/inc/us/HttpServer.hpp b/libhttp/inc/us/HttpServer.hpp index f157f38..843bd20 100644 --- a/libhttp/inc/us/HttpServer.hpp +++ b/libhttp/inc/us/HttpServer.hpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -19,12 +20,14 @@ #include #include #include +#include namespace pmc { namespace net { namespace beast = boost::beast; namespace http = beast::http; +namespace websocket = beast::websocket; namespace asio = boost::asio; using tcp = boost::asio::ip::tcp; @@ -37,6 +40,19 @@ using MiddlewareHandler = std::function&, const std::unordered_map&)>; +// WebSocket 回调类型 +using WebSocketMessageHandler = std::function sendCallback)>; + +using WebSocketConnectHandler = std::function; + +using WebSocketDisconnectHandler = std::function; + struct RouteEntry { std::string path; std::regex pattern; @@ -44,6 +60,11 @@ struct RouteEntry { bool is_wildcard; }; +struct WsRouteEntry { + std::string path; + WebSocketMessageHandler handler; +}; + class HttpServer { public: explicit HttpServer(const std::string& address, unsigned short port, @@ -60,10 +81,19 @@ public: void post(const std::string& path, HttpRequestHandler handler); void put(const std::string& path, HttpRequestHandler handler); void del(const std::string& path, HttpRequestHandler handler); - void options(const std::string& path, HttpRequestHandler handler); // 新增 OPTIONS + void options(const std::string& path, HttpRequestHandler handler); void use(MiddlewareHandler middleware); void setStaticDirectory(const std::string& path); + // WebSocket 接口 + void ws(const std::string& path, WebSocketMessageHandler handler); + void onWsConnect(WebSocketConnectHandler handler); + void onWsDisconnect(WebSocketDisconnectHandler handler); + bool sendToWsClient(const std::string& clientId, const std::string& message); + size_t broadcastWs(const std::string& message); + void disconnectWsClient(const std::string& clientId); + size_t getWsClientCount() const; + bool isRunning() const; unsigned short getPort() const; std::string getAddress() const; @@ -103,10 +133,18 @@ private: std::vector post_handlers_; std::vector put_handlers_; std::vector delete_handlers_; - std::vector options_handlers_; // 新增 OPTIONS 路由 + std::vector options_handlers_; std::vector middlewares_; std::string static_directory_; + // WebSocket 路由和处理 + std::vector ws_handlers_; + WebSocketConnectHandler ws_connect_handler_; + WebSocketDisconnectHandler ws_disconnect_handler_; + std::unordered_map> ws_clients_; + mutable std::mutex ws_clients_mutex_; + std::atomic ws_client_counter_{0}; + std::vector> sessions_; std::mutex sessions_mutex_; @@ -115,12 +153,21 @@ private: bool matchPath(const std::string& path, const std::regex& pattern); http::response handleRequest(const http::request& req); HttpRequestHandler findRoute(const std::string& method, const std::string& path); - bool routeExists(const std::string& method, const std::string& path); // 新增辅助方法 + bool routeExists(const std::string& method, const std::string& path); + WebSocketMessageHandler findWsRoute(const std::string& path); + + // WebSocket 客户端管理 + std::string generateWsClientId(); + void addWsClient(const std::string& clientId, std::shared_ptr session); + void removeWsClient(const std::string& clientId); + std::shared_ptr getWsClient(const std::string& clientId); void addSession(const std::shared_ptr& session); void removeSession(const Session* session); void closeAllSessions(); void initSSLContext(); + + friend class Session; }; } // namespace net diff --git a/libhttp/src/HttpServer.cpp b/libhttp/src/HttpServer.cpp index 588cad8..a918262 100644 --- a/libhttp/src/HttpServer.cpp +++ b/libhttp/src/HttpServer.cpp @@ -5,6 +5,7 @@ #include #include #include +#include namespace pmc { namespace net { @@ -29,25 +30,30 @@ public: // ===================== 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; + public: - // HTTP 构造 Session(tcp::socket&& socket, HttpServer* server) - : socket_(std::move(socket)), server_(server), ssl_enabled_(false), - strand_(asio::make_strand(socket_.get_executor())) {} + : stream_(std::move(socket)), server_(server), ssl_enabled_(false), + is_websocket_(false), client_id_("") {} - // HTTPS 构造 Session(tcp::socket&& socket, asio::ssl::context& ssl_ctx, HttpServer* server) - : socket_(std::move(socket)), server_(server), ssl_enabled_(true), - strand_(asio::make_strand(socket_.get_executor())) { - ssl_stream_ = std::make_unique>( - std::move(socket_), ssl_ctx); + : stream_(std::move(socket)), server_(server), ssl_enabled_(true), + is_websocket_(false), client_id_("") { + ssl_stream_ = std::make_unique(stream_, ssl_ctx); } ~Session() { close(); } void run() { - if (ssl_enabled_) doSSLHandshake(); - else doRead(); + if (ssl_enabled_) { + doSSLHandshake(); + } else { + doRead(); + } } void setRemoteInfo(const std::string& ip, unsigned short port) { @@ -55,16 +61,35 @@ public: 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; - if (ssl_enabled_ && ssl_stream_) { - ssl_stream_->shutdown(ec); - ssl_stream_->lowest_layer().close(ec); - } else if (socket_.is_open()) { - socket_.shutdown(tcp::socket::shutdown_send, ec); - socket_.close(ec); + 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() { @@ -74,39 +99,60 @@ private: ssl_stream_->async_handshake( asio::ssl::stream_base::server, - asio::bind_executor(strand_, - [this, self](beast::error_code ec) { - if (!ec) { - std::cout << "[SSL] Handshake succeeded: " - << remote_ip_ << ":" << remote_port_ << std::endl; - doRead(); - } else { - std::cerr << "[SSL] Handshake error: " - << remote_ip_ << ":" << remote_port_ - << " - " << ec.message() - << " (code=" << ec.value() << ")" << std::endl; - ERR_print_errors_fp(stderr); - } - })); + [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_, - asio::bind_executor(strand_, - [this, self](beast::error_code ec, std::size_t) { handleRead(ec); })); + [self](beast::error_code ec, std::size_t) { + self->handleRead(ec); + }); } else { - http::async_read(socket_, buffer_, request_, - asio::bind_executor(strand_, - [this, self](beast::error_code ec, std::size_t) { handleRead(ec); })); + 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); + + 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_); - request_ = {}; processRequest(std::move(req_copy)); } else if (ec != http::error::end_of_stream) { std::cerr << "[Read] Error from " << remote_ip_ << ":" << remote_port_ @@ -114,6 +160,154 @@ private: } } + 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 (server_->ws_connect_handler_) { + server_->ws_connect_handler_(client_id_, getRemoteAddress()); + } + + doWsRead(); + } + + 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; + + server_->removeWsClient(client_id_); + + if (server_->ws_disconnect_handler_) { + server_->ws_disconnect_handler_(client_id_); + } + client_id_.clear(); + is_websocket_ = false; + } + void processRequest(http::request req) { try { auto resp = std::make_shared>( @@ -135,42 +329,49 @@ private: auto self = shared_from_this(); if (ssl_enabled_ && ssl_stream_) { http::async_write(*ssl_stream_, *response, - asio::bind_executor(strand_, - [this, self, response](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[Write] Error to " << remote_ip_ << ":" << remote_port_ - << ": " << ec.message() << std::endl; - return; - } - if (!response->keep_alive()) close(); - else doRead(); - })); + [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(socket_, *response, - asio::bind_executor(strand_, - [this, self, response](beast::error_code ec, std::size_t) { - if (ec) { - std::cerr << "[Write] Error to " << remote_ip_ << ":" << remote_port_ - << ": " << ec.message() << std::endl; - return; - } - if (!response->keep_alive()) close(); - else doRead(); - })); + 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(); + }); } } - tcp::socket socket_; - std::unique_ptr> ssl_stream_; + 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_; - asio::strand strand_; - std::string remote_ip_ = "unknown"; unsigned short remote_port_ = 0; + + bool is_websocket_; + std::string client_id_; + std::string ws_path_; + WebSocketMessageHandler ws_handler_; + std::deque ws_send_queue_; }; // ===================== Listener ===================== @@ -204,8 +405,10 @@ private: } void doAccept() { + auto self = shared_from_this(); acceptor_.async_accept( - [this, self = shared_from_this()](beast::error_code ec, tcp::socket socket) { + 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(); @@ -213,17 +416,18 @@ private: std::cout << "[Listener] New connection from " << ip << ":" << port << std::endl; std::shared_ptr session; - if (ssl_enabled_ && ssl_ctx_) { - session = std::make_shared(std::move(socket), *ssl_ctx_, server_); + 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), server_); + 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; } - doAccept(); + self->doAccept(); }); } @@ -236,7 +440,6 @@ private: // ===================== HttpServer 实现 ===================== -// ---- 构造函数 ---- HttpServer::HttpServer(const std::string& address, unsigned short port, unsigned int threads) : address_(address), port_(port), threads_(threads), ssl_enabled_(false), ioc_(std::make_unique(threads)) { @@ -255,7 +458,6 @@ HttpServer::HttpServer(const std::string& address, unsigned short port, HttpServer::~HttpServer() { stop(); } -// ---- SSL 初始化 ---- void HttpServer::setPrivateKeyPassword(const std::string& password) { private_key_password_ = password; } @@ -294,7 +496,6 @@ void HttpServer::initSSLContext() { } } -// ---- 启动/停止 ---- void HttpServer::start() { if (running_) return; try { @@ -318,7 +519,11 @@ void HttpServer::start() { } std::cout << (ssl_enabled_ ? "HTTPS" : "HTTP") << " Server started on " << address_ << ":" << port_ - << " with " << threads_ << " threads\n"; + << " with " << threads_ << " threads"; + if (!ws_handlers_.empty()) { + std::cout << " (WebSocket enabled, " << ws_handlers_.size() << " WS route(s))"; + } + std::cout << "\n"; } catch (const std::exception& e) { std::cerr << "[Start] Failed: " << e.what() << std::endl; running_ = false; @@ -328,6 +533,12 @@ void HttpServer::start() { void HttpServer::stop() { if (!running_) return; running_ = false; + + { + std::lock_guard lock(ws_clients_mutex_); + ws_clients_.clear(); + } + closeAllSessions(); if (ioc_) ioc_->stop(); for (auto& t : worker_threads_) if (t.joinable()) t.join(); @@ -341,13 +552,12 @@ void HttpServer::run() { if (ioc_) ioc_->run(); } -// ---- 状态查询 ---- bool HttpServer::isRunning() const { return running_; } 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; @@ -388,7 +598,6 @@ void HttpServer::del(const std::string& path, HttpRequestHandler handler) { delete_handlers_.push_back(entry); } -// ---- 新增 OPTIONS 路由注册 ---- void HttpServer::options(const std::string& path, HttpRequestHandler handler) { RouteEntry entry; entry.path = path; @@ -407,6 +616,62 @@ 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::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; + } + } + 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; @@ -453,22 +718,48 @@ HttpRequestHandler HttpServer::findRoute(const std::string& method, const std::s 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; +} + +// ---- 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()); @@ -486,7 +777,6 @@ http::response HttpServer::handleRequest(const http::request< 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)) { @@ -500,10 +790,8 @@ http::response HttpServer::handleRequest(const http::request< } std::string method = http::to_string(req.method()); - - // ========== 处理 OPTIONS 预检请求 ========== + if (method == "OPTIONS") { - // 检查是否有自定义 OPTIONS 路由 auto handler = findRoute("OPTIONS", target_path); if (handler) { try { @@ -516,34 +804,33 @@ http::response HttpServer::handleRequest(const http::request< return res; } } - - // 自动生成 OPTIONS 响应 + 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()) { + + 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, + 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, + 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() = ""; @@ -552,19 +839,15 @@ http::response HttpServer::handleRequest(const http::request< return res; } } - - // 没有 Origin 头或没有匹配的方法 + res.result(http::status::not_found); res.body() = "404 Not Found\n"; res.prepare_payload(); return res; } - // ========== OPTIONS 处理结束 ========== - // 查找普通路由 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); @@ -592,7 +875,6 @@ http::response HttpServer::handleRequest(const http::request< } } - // 路由处理 if (handler) { try { return handler(req, params); @@ -605,14 +887,12 @@ http::response HttpServer::handleRequest(const http::request< } } - // 404 res.result(http::status::not_found); res.body() = "404 Not Found\n"; res.prepare_payload(); return res; } -// ---- 状态显示 ---- void HttpServer::showStatus() const { std::cout << "=== " << (ssl_enabled_ ? "HTTPS" : "HTTP") << " Server Status ===" << std::endl; std::cout << "Protocol: " << (ssl_enabled_ ? "HTTPS (SSL/TLS)" : "HTTP") << std::endl; @@ -638,9 +918,13 @@ void HttpServer::showStatus() const { std::cout << " DELETE: " << delete_handlers_.size() << " handler(s)" << std::endl; for (const auto& entry : delete_handlers_) std::cout << " DELETE " << entry.path << (entry.is_wildcard ? " (wildcard)" : "") << std::endl; - std::cout << " OPTIONS: " << options_handlers_.size() << " handler(s)" << std::endl; // 新增 + std::cout << " OPTIONS: " << options_handlers_.size() << " handler(s)" << std::endl; for (const auto& entry : options_handlers_) std::cout << " OPTIONS " << entry.path << (entry.is_wildcard ? " (wildcard)" : "") << std::endl; + std::cout << " WebSocket: " << ws_handlers_.size() << " route(s)" << std::endl; + for (const auto& entry : ws_handlers_) + std::cout << " WS " << entry.path << std::endl; + std::cout << " WebSocket clients: " << getWsClientCount() << std::endl; std::cout << "Middlewares: " << middlewares_.size() << std::endl; std::cout << "==========================" << std::endl; } @@ -664,10 +948,6 @@ void HttpServer::removeSession(const Session* session) { void HttpServer::closeAllSessions() { std::lock_guard lock(sessions_mutex_); - for (auto& weak : sessions_) { - auto sp = weak.lock(); - if (sp) { /* 会话析构时自动关闭 socket */ } - } sessions_.clear(); } diff --git a/libhttp/test_server.cpp b/libhttp/test_server.cpp new file mode 100644 index 0000000..db400f6 --- /dev/null +++ b/libhttp/test_server.cpp @@ -0,0 +1,145 @@ +#include "us/HttpServer.hpp" +#include +#include +#include +#include + +using namespace pmc::net; + +static std::unique_ptr g_server; + +void signalHandler(int signum) { + std::cout << "\nStopping server..." << std::endl; + if (g_server) g_server->stop(); +} + +int main() { + signal(SIGINT, signalHandler); + signal(SIGTERM, signalHandler); + + try { + // 创建 HTTP 服务器（同时支持 HTTP 和 WebSocket） + g_server = std::make_unique("0.0.0.0", 9090, 2); + + // 注册普通 HTTP GET 接口 + g_server->get("/hello", [](const auto& req, const auto& params) { + http::response res; + res.version(req.version()); + res.set(http::field::content_type, "text/plain; charset=utf-8"); + res.result(http::status::ok); + res.body() = "Hello from HTTP! Server is working correctly."; + res.prepare_payload(); + return res; + }); + + // 注册 HTTP POST 接口 + g_server->post("/echo", [](const auto& req, const auto& params) { + http::response res; + res.version(req.version()); + res.set(http::field::content_type, "application/json; charset=utf-8"); + res.result(http::status::ok); + res.body() = std::string("{\"received\":\"") + req.body() + "\"}"; + res.prepare_payload(); + return res; + }); + + // WebSocket 测试页面 + g_server->get("/test", [](const auto& req, const auto& params) { + http::response res; + res.version(req.version()); + res.set(http::field::content_type, "text/html; charset=utf-8"); + res.result(http::status::ok); + res.body() = R"HTML( +WebSocket Test + + +

HTTP+WebSocket Integration Test

+

HTTP: /hello | POST /echo

+ + + +
+)HTML"; + res.prepare_payload(); + return res; + }); + + // 注册 WebSocket 接口 - Echo 服务器 + g_server->ws("/ws", [](const std::string& clientId, const std::string& message, + std::function send) { + std::cout << "[WS] Received from " << clientId << ": " << message << std::endl; + // Echo back with prefix + send(std::string("Echo: ") + message); + }); + + // WebSocket 另一个接口 - Chat + g_server->ws("/chat", [](const std::string& clientId, const std::string& message, + std::function send) { + std::cout << "[Chat] Received from " << clientId << ": " << message << std::endl; + send(std::string("[Chat] ") + message); + }); + + // WebSocket 连接回调 + g_server->onWsConnect([](const std::string& clientId, const std::string& address) { + std::cout << "[WS] Client connected: " << clientId << " from " << address << std::endl; + }); + + // WebSocket 断开回调 + g_server->onWsDisconnect([](const std::string& clientId) { + std::cout << "[WS] Client disconnected: " << clientId << std::endl; + }); + + // 启动服务器 + g_server->showStatus(); + g_server->start(); + + std::cout << "\n=== Test Server Running ===" << std::endl; + std::cout << "HTTP test: curl http://localhost:9090/hello" << std::endl; + std::cout << "WebSocket test: ws://localhost:9090/ws (echo) or ws://localhost:9090/chat (chat)" << std::endl; + std::cout << "Press Ctrl+C to stop" << std::endl; + std::cout << std::endl; + + // 3秒后广播一条消息给所有 WebSocket 客户端 + std::thread broadcast_thread([&]() { + std::this_thread::sleep_for(std::chrono::seconds(3)); + if (g_server && g_server->getWsClientCount() > 0) { + std::cout << "Broadcasting to all clients..." << std::endl; + g_server->broadcastWs("Server broadcast after 3 seconds!"); + } + }); + broadcast_thread.detach(); + + // 运行事件循环 + g_server->run(); + + } catch (const std::exception& e) { + std::cerr << "Server error: " << e.what() << std::endl; + return 1; + } + + std::cout << "Server stopped." << std::endl; + return 0; +}