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

@@ -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))
 

libhttp/inc/us/HttpServer.hpp

@@ -4,6 +4,7 @@
 #include <boost/asio.hpp>
 #include <boost/beast.hpp>
 #include <boost/beast/http.hpp>
+#include <boost/beast/websocket.hpp>
 #include <boost/beast/version.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/ssl.hpp>
@@ -19,12 +20,14 @@
 #include <atomic>
 #include <mutex>
 #include <regex>
+#include <deque>
 
 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<bool(
     http::response<http::string_body>&,
     const std::unordered_map<std::string, std::string>&)>;
 
+// WebSocket 回调类型
+using WebSocketMessageHandler = std::function<void(
+    const std::string& clientId,
+    const std::string& message,
+    std::function<void(const std::string&)> sendCallback)>;
+
+using WebSocketConnectHandler = std::function<void(
+    const std::string& clientId,
+    const std::string& remoteAddress)>;
+
+using WebSocketDisconnectHandler = std::function<void(
+    const std::string& clientId)>;
+
 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<RouteEntry> post_handlers_;
     std::vector<RouteEntry> put_handlers_;
     std::vector<RouteEntry> delete_handlers_;
-    std::vector<RouteEntry> options_handlers_;  // 新增 OPTIONS 路由
+    std::vector<RouteEntry> options_handlers_;
     std::vector<MiddlewareHandler> middlewares_;
     std::string static_directory_;
 
+    // WebSocket 路由和处理
+    std::vector<WsRouteEntry> ws_handlers_;
+    WebSocketConnectHandler ws_connect_handler_;
+    WebSocketDisconnectHandler ws_disconnect_handler_;
+    std::unordered_map<std::string, std::shared_ptr<Session>> ws_clients_;
+    mutable std::mutex ws_clients_mutex_;
+    std::atomic<uint64_t> ws_client_counter_{0};
+
     std::vector<std::weak_ptr<Session>> sessions_;
     std::mutex sessions_mutex_;
 
@@ -115,12 +153,21 @@ private:
     bool matchPath(const std::string& path, const std::regex& pattern);
     http::response<http::string_body> handleRequest(const http::request<http::string_body>& 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> session);
+    void removeWsClient(const std::string& clientId);
+    std::shared_ptr<Session> getWsClient(const std::string& clientId);
 
     void addSession(const std::shared_ptr<Session>& session);
     void removeSession(const Session* session);
     void closeAllSessions();
     void initSSLContext();
+
+    friend class Session;
 };
 
 } // namespace net

libhttp/src/HttpServer.cpp

@@ -5,6 +5,7 @@
 #include <iomanip>
 #include <openssl/err.h>
 #include <openssl/ssl.h>
+#include <boost/beast/websocket/ssl.hpp>
 
 namespace pmc {
 namespace net {
@@ -29,25 +30,30 @@ public:
 
 // ===================== Session =====================
 class HttpServer::Session : public std::enable_shared_from_this<Session> {
+    using tcp_stream = beast::tcp_stream;
+    using ssl_stream = asio::ssl::stream<tcp_stream&>;
+    using ws_stream = websocket::stream<tcp_stream&>;
+    using wss_stream = websocket::stream<ssl_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<asio::ssl::stream<tcp::socket>>(
-            std::move(socket_), ssl_ctx);
+        : stream_(std::move(socket)), server_(server), ssl_enabled_(true),
+          is_websocket_(false), client_id_("") {
+        ssl_stream_ = std::make_unique<ssl_stream>(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<http::string_body>& 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<ws_stream>(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<wss_stream>(*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<http::string_body> req) {
         try {
             auto resp = std::make_shared<http::response<http::string_body>>(
@@ -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<asio::ssl::stream<tcp::socket>> ssl_stream_;
+    tcp_stream stream_;
+    std::unique_ptr<ssl_stream> ssl_stream_;
+    std::unique_ptr<ws_stream>  ws_stream_;
+    std::unique_ptr<wss_stream> wss_stream_;
+
     beast::flat_buffer buffer_;
+    beast::flat_buffer ws_buffer_;
     http::request<http::string_body> request_;
+
     HttpServer* server_;
     bool ssl_enabled_;
 
-    asio::strand<asio::any_io_executor> 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<std::string> 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> session;
-                    if (ssl_enabled_ && ssl_ctx_) {
-                        session = std::make_shared<Session>(std::move(socket), *ssl_ctx_, server_);
+                    if (self->ssl_enabled_ && self->ssl_ctx_) {
+                        session = std::make_shared<Session>(std::move(socket), *self->ssl_ctx_, self->server_);
                     } else {
-                        session = std::make_shared<Session>(std::move(socket), server_);
+                        session = std::make_shared<Session>(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<asio::io_context>(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<std::mutex> 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> session;
+    {
+        std::lock_guard<std::mutex> 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<std::shared_ptr<Session>> clients;
+    {
+        std::lock_guard<std::mutex> 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<std::mutex> lock(ws_clients_mutex_);
+    ws_clients_.erase(clientId);
+}
+
+size_t HttpServer::getWsClientCount() const {
+    std::lock_guard<std::mutex> 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> session) {
+    std::lock_guard<std::mutex> lock(ws_clients_mutex_);
+    ws_clients_[clientId] = session;
+}
+
+void HttpServer::removeWsClient(const std::string& clientId) {
+    std::lock_guard<std::mutex> lock(ws_clients_mutex_);
+    ws_clients_.erase(clientId);
+}
+
+std::shared_ptr<HttpServer::Session> HttpServer::getWsClient(const std::string& clientId) {
+    std::lock_guard<std::mutex> lock(ws_clients_mutex_);
+    auto it = ws_clients_.find(clientId);
+    if (it != ws_clients_.end()) return it->second;
+    return nullptr;
+}
+
 // ---- 请求处理核心 ----
 http::response<http::string_body> HttpServer::handleRequest(const http::request<http::string_body>& req) {
     std::string target_path(req.target().data(), req.target().size());
@@ -486,7 +777,6 @@ http::response<http::string_body> 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<http::string_body> 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<http::string_body> HttpServer::handleRequest(const http::request<
                 return res;
             }
         }
-        
-        // 自动生成 OPTIONS 响应
+
         auto origin = req.find(http::field::origin);
         if (origin != req.end()) {
-            // 构建允许的方法列表
             std::vector<std::string> all_methods = {"GET", "POST", "PUT", "DELETE", "OPTIONS"};
             std::vector<std::string> 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<http::string_body> 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<http::string_body> HttpServer::handleRequest(const http::request<
         }
     }
 
-    // 路由处理
     if (handler) {
         try {
             return handler(req, params);
@@ -605,14 +887,12 @@ http::response<http::string_body> 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<std::mutex> lock(sessions_mutex_);
-    for (auto& weak : sessions_) {
-        auto sp = weak.lock();
-        if (sp) { /* 会话析构时自动关闭 socket */ }
-    }
     sessions_.clear();
 }
 

libhttp/test_server.cpp

@@ -0,0 +1,145 @@
+#include "us/HttpServer.hpp"
+#include <iostream>
+#include <thread>
+#include <chrono>
+#include <signal.h>
+
+using namespace pmc::net;
+
+static std::unique_ptr<HttpServer> 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<HttpServer>("0.0.0.0", 9090, 2);
+
+        // 注册普通 HTTP GET 接口
+        g_server->get("/hello", [](const auto& req, const auto& params) {
+            http::response<http::string_body> 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<http::string_body> 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<http::string_body> 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(<!DOCTYPE html>
+<html><head><title>WebSocket Test</title>
+<style>body{font-family:monospace;padding:20px}#log{border:1px solid #ccc;padding:10px;height:300px;overflow-y:scroll}</style>
+</head><body>
+<h2>HTTP+WebSocket Integration Test</h2>
+<p>HTTP: <a href="/hello">/hello</a> | <a href="#" onclick="testPost();return false">POST /echo</a></p>
+<button onclick="testWs('/ws')">Connect /ws (echo)</button>
+<button onclick="testWs('/chat')">Connect /chat</button>
+<button onclick="clearLog()">Clear</button>
+<div id="log"></div>
+<script>
+const logDiv = document.getElementById('log');
+function log(msg, color) {
+  color = color || 'black';
+  logDiv.innerHTML += '<span style="color:'+color+'">'+new Date().toLocaleTimeString()+' - '+msg+'</span><br>';
+  logDiv.scrollTop = logDiv.scrollHeight;
+}
+function clearLog() { logDiv.innerHTML = ''; }
+
+function testPost() {
+  fetch('/echo',{method:'POST',body:'Hello POST'})
+    .then(function(r){return r.text()}).then(function(t){log('POST /echo: '+t,'blue')});
+}
+
+function testWs(path) {
+  log('Connecting to ws://'+location.host+path+'...');
+  var ws = new WebSocket('ws://'+location.host+path);
+  ws.onopen = function() { log(path+' connected','green'); ws.send('Hello from browser to '+path); };
+  ws.onmessage = function(e) { log(path+' received: '+e.data,'blue'); };
+  ws.onerror = function(e) { log(path+' error','red'); };
+  ws.onclose = function() { log(path+' closed','gray'); };
+}
+log('Page loaded. HTTP server working. Use buttons above to test WebSocket.');
+</script></body></html>)HTML";
+            res.prepare_payload();
+            return res;
+        });
+
+        // 注册 WebSocket 接口 - Echo 服务器
+        g_server->ws("/ws", [](const std::string& clientId, const std::string& message,
+                               std::function<void(const std::string&)> 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<void(const std::string&)> 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;
+}