websocket.hpp

浏览该文件的文档.

#ifndef NEFORCE_NETWORK_HTTP_WEBSOCKET_HPP__
#define NEFORCE_NETWORK_HTTP_WEBSOCKET_HPP__

 
#include "NeForce/core/async/atomic.hpp"
#include "NeForce/core/async/condition_variable.hpp"
#include "NeForce/core/async/io_context.hpp"
#include "NeForce/core/async/thread.hpp"
#include "NeForce/core/container/queue.hpp"
#include "NeForce/core/container/unordered_map.hpp"
#include "NeForce/core/functional/function.hpp"
#include "NeForce/core/memory/weak_ptr.hpp"
#include "NeForce/network/http/http_server_message.hpp"
#include "NeForce/network/http/websocket_deflate.hpp"
#include "NeForce/network/ssl/ssl_socket.hpp"
NEFORCE_BEGIN_NAMESPACE__
NEFORCE_BEGIN_HTTP__



enum class websocket_status : uint16_t {
    NORMAL_CLOSURE = 1000,             
    GOING_AWAY = 1001,                 
    PROTOCOL_ERROR = 1002,             
    UNSUPPORTED_DATA = 1003,           
    RESERVED = 1004,                   
    NO_STATUS_RCVD = 1005,             
    ABNORMAL_CLOSURE = 1006,           
    INVALID_FRAME_PAYLOAD_DATA = 1007, 
    POLICY_VIOLATION = 1008,           
    MESSAGE_TOO_BIG = 1009,            
    MANDATORY_EXT = 1010,              
    INTERNAL_ERROR = 1011,             
    SERVICE_RESTART = 1012,            
    TRY_AGAIN_LATER = 1013,            
    BAD_GATEWAY = 1014,                
    TLS_HANDSHAKE = 1015               
};

enum class websocket_opcode : uint8_t {
    CONTINUATION = 0x0, 
    TEXT = 0x1,         
    BINARY = 0x2,       
    CLOSE = 0x8,        
    PING = 0x9,         
    PONG = 0xA          
};
 
#pragma pack(push, 1)
struct websocket_frame_header {
    byte_t opcode : 4;      
    byte_t rsv3 : 1;        
    byte_t rsv2 : 1;        
    byte_t rsv1 : 1;        
    byte_t fin : 1;         
    byte_t payload_len : 7; 
    byte_t masked : 1;      
};
#pragma pack(pop)
 
class websocket_session;
 

class NEFORCE_API websocket_server {
public:
    using session_ptr = shared_ptr<websocket_session>;   
    using session_handler = function<void(session_ptr)>; 
 
private:
    unordered_map<string, session_handler> route_handlers_; 
    vector<session_ptr> sessions_;                          
    mutable mutex sessions_mutex_;                          
    io_context* ctx_{nullptr};                              
 
public:
    websocket_server() = default;
    ~websocket_server();
 
    websocket_server(const websocket_server&) = delete;
    websocket_server& operator=(const websocket_server&) = delete;
 
    websocket_server(websocket_server&&) noexcept = delete;
    websocket_server& operator=(websocket_server&&) noexcept = delete;

    void set_io_context(io_context& ctx) noexcept { ctx_ = &ctx; }

    void route(const string& path, session_handler handler) { route_handlers_[path] = _NEFORCE move(handler); }

    bool handle_upgrade(const http_request& request, unique_ptr<tcp_socket> sock);

    void remove_session(const session_ptr& session);
    void stop();

    void broadcast(const string& data, websocket_opcode opcode = websocket_opcode::TEXT);

    size_t session_count() const noexcept {
        lock<mutex> lk(sessions_mutex_);
        return sessions_.size();
    }
 
    // TODO: STOMP sub-protocol support — STOMP 1.2 over WebSocket with destination-based routing, subscription management, ACK modes
    // TODO: SockJS fallback — HTTP long-polling / XHR streaming fallback for browsers without WebSocket support
};

class NEFORCE_API websocket_session : public enable_shared_from_this<websocket_session> {
public:
    using message_handler = function<void(const string&, websocket_opcode)>; 
    using close_handler = function<void(websocket_status, const string&)>;   
    using error_handler = function<void(const exception&)>;                  
 
private:
    unique_ptr<tcp_socket> socket_; 
    websocket_server* server_;      
 
    atomic<bool> running_{false}; 
    atomic_flag closed_once_;     
 
    thread read_thread_;      
    thread write_thread_;     
    thread heartbeat_thread_; 
 
    io_context* ctx_{nullptr};      
    bool event_driven_ = false;     
    size_t heartbeat_timer_id_ = 0; 
    byte_vector read_buffer_;       
    bool write_registered_ = false; 
 
    mutex write_mutex_;              
    condition_variable write_cv_;    
    queue<byte_vector> write_queue_; 
    queue<byte_vector> ctrl_queue_;  
 
    string fragment_buffer_;                                    
    websocket_opcode fragment_opcode_ = websocket_opcode::TEXT; 
    bool in_fragment_ = false;                                  
 
    atomic<bool> ping_pending_{false}; 
    atomic<int64_t> last_pong_ms_{0};  
 
    websocket_deflate_config deflate_config_;            
    unique_ptr<websocket_deflate> deflate_compressor_;   
    unique_ptr<websocket_deflate> deflate_decompressor_; 
    byte_vector deflate_fragment_buffer_;                
 
    message_handler on_message_; 
    close_handler on_close_;     
    error_handler on_error_;     
 
    bool queue_frame(byte_vector frame, bool is_control = false);
    void write_loop();
 
    void read_loop();
    bool read_frame();
 
    bool dispatch(const websocket_frame_header& hdr, websocket_opcode opcode, string payload);
    void deliver_message(const string& data, websocket_opcode opcode);
 
    void send_close_frame(websocket_status status, const string& reason);
    void handle_close_frame(string payload);
 
    void heartbeat_loop();
    void do_stop(websocket_status status, const string& reason, bool notify_server = true);
 
    void start_event_driven();
 
    void on_readable(int fd, uint32_t events, error_code ec);
    void on_writable(int fd, uint32_t events, error_code ec);
    void on_heartbeat_timer();
 
    void flush_event_writes();
    void try_parse_frames();
 
public:
    explicit websocket_session(unique_ptr<tcp_socket> sock, websocket_server* server = nullptr);

    ~websocket_session();
 
    websocket_session(const websocket_session&) = delete;
    websocket_session& operator=(const websocket_session&) = delete;

    void start();

    void close(websocket_status status = websocket_status::NORMAL_CLOSURE, const string& reason = "");

    void stop();

    bool send(const string& data, websocket_opcode opcode = websocket_opcode::TEXT);

    bool send_binary(const string& data) { return send(data, websocket_opcode::BINARY); }

    bool is_open() const noexcept { return running_ && socket_->is_open(); }

    void set_message_handler(message_handler handler) { on_message_ = _NEFORCE move(handler); }

    void set_close_handler(close_handler handler) { on_close_ = _NEFORCE move(handler); }

    void set_error_handler(error_handler handler) { on_error_ = _NEFORCE move(handler); }

    void set_deflate_config(const websocket_deflate_config& cfg);

    const websocket_deflate_config& deflate_config() const noexcept { return deflate_config_; }

    bool has_deflate_config() const noexcept { return deflate_config_.active; }

    void set_io_context(io_context& ctx) noexcept {
        ctx_ = &ctx;
        event_driven_ = true;
    }

    tcp_socket& socket() noexcept { return *socket_; }

    const tcp_socket& socket() const noexcept { return *socket_; }

    ssl_socket* ssl_socket_ptr() noexcept { return dynamic_cast<ssl_socket*>(socket_.get()); }

    const ssl_socket* ssl_socket_ptr() const noexcept { return dynamic_cast<const ssl_socket*>(socket_.get()); }
};
 // WebSocket
 // HTTP
 
NEFORCE_END_HTTP__
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_NETWORK_HTTP_WEBSOCKET_HPP__