eiennet_websocket.hpp

浏览该文件的文档.

#pragma once

namespace eiennet
{

namespace ws
{
    /*
        一个为未分片的消息（FIN为1，opcode非0）
        一个分片的消息由起始帧：
        （FIN为0，opcode非0），
        若干（0个或多个）帧（FIN为0，opcode为0），
        结束帧（FIN为1，opcode为0）
    */
    /*
        HTTP升级为websocket连接的请求
        GET /xxx HTTP/1.1
        Host: hostname[:port]
        Upgrade: websocket
        Connection: Upgrade
        Sec-WebSocket-Key: 随机生成的16字节的值用BASE64编码后的字符串
        Origin: 其值为请求发起页面的scheme://hostname (因为中间可能经过代理才到达服务器，服务器可以根据该字段选择是否和客户端建立连接)
        Sec-WebSocket-Version: 13
        [Sec-WebSocket-Protocol: 其值为由逗号分隔的子协议的名字，按优先度排序，每个名字必须唯一]
        [Sec-WebSocket-Extension: 表示协议级别的扩展]

        连接建立的合法响应
        HTTP/1.1 101 xxx
        Upgrade: websocket
        Connection: Upgrade
        Sec-WebSocket-Accept: 其值为 Base64( SHA1( RequestHeader{Sec-WebSocket-Key} + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" ) )
        [Sec-WebSocket-Protocol: 其值为请求发来的子协议中的一个或空]
    */

enum OpCode : winux::uint
{
    dataContinued = 0x0U, 
    dataText = 0x1U, 
    dataBinary = 0x2U, 
    dataRsv3 = 0x3U,
    dataRsv4 = 0x4U,
    dataRsv5 = 0x5U,
    dataRsv6 = 0x6U,
    dataRsv7 = 0x7U,
    ctrlClose = 0x8U, 
    ctrlPing = 0x9U, 
    ctrlPong = 0xAU, 
    ctrlRsvB = 0xBU,
    ctrlRsvC = 0xCU,
    ctrlRsvD = 0xDU,
    ctrlRsvE = 0xEU,
    ctrlRsvF = 0xFU,
};

enum WebSocketErrorCode
{
    errNothing,
    errCnnUnexpectedInvalid,
    //errCnnUpgradeFail,
};

enum FrameType
{
    ftBasic = 0b0010,
    ftBasicMasked = 0b0011,
    ftExtended16 = 0b0100,
    ftExtended16Masked = 0b0101,
    ftExtended64 = 0b1000,
    ftExtended64Masked = 0b1001,
};

enum ReadyState
{
    stateConnecting,
    stateOpen,
    stateClosing,
    stateClosed
};

#pragma pack(push, 1)

struct FrameBase
{
    winux::uint8 opcode:4;
    winux::uint8 rsv3:1;
    winux::uint8 rsv2:1;
    winux::uint8 rsv1:1;
    winux::uint8 fin:1;

    winux::uint8 payloadLen:7;
    winux::uint8 mask:1;

    FrameBase()
    {
        memset( this, 0, sizeof(*this) );
    }
};

// WebSocket数据帧根据Payload-len来决定帧头部大小
// Payload-len < 126: 2bytes, 16bits
// Payload-len ==126: 4bytes, 32bits
// Payload-len ==127: 10bytes, 80bits
// 除此之外，如果mask位为1，则包含32bits的掩码Key,即
// Payload-len < 126: 6bytes, 48bits
// Payload-len ==126: 8bytes, 64bits
// Payload-len ==127: 14bytes, 112bits

// 基本帧 payload len < 126, mask == 0
struct FrameBasic : FrameBase
{
    FrameBasic()
    {
        memset( this, 0, sizeof(*this) );
    }

    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};
#pragma pack(pop)

#pragma pack(push, 2)

// 基本掩码帧 payload len < 126, mask == 1
struct FrameBasicMasked : FrameBase
{
    winux::uint8 maskingKey[4];

    FrameBasicMasked()
    {
        memset( this, 0, sizeof(*this) );
    }

    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};

// 16位数据扩展帧 payload len == 126, mask == 0
struct FrameExtended16 : FrameBase
{
    winux::uint16 extendedPayloadLen;

    FrameExtended16()
    {
        memset( this, 0, sizeof(*this) );
    }

    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};

// 16位数据扩展掩码帧 payload len == 126, mask == 1
struct FrameExtended16Masked : FrameBase
{
    winux::uint16 extendedPayloadLen;
    winux::uint8 maskingKey[4];

    FrameExtended16Masked()
    {
        memset( this, 0, sizeof(*this) );
    }

    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};

// 64位数据扩展帧 payload len == 127, mask == 0
struct FrameExtended64 : FrameBase
{
    winux::uint64 extendedPayloadLen;

    FrameExtended64()
    {
        memset( this, 0, sizeof(*this) );
    }

    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};

// 64位数据扩展掩码帧 payload len == 127, mask == 1
struct FrameExtended64Masked : FrameBase
{
    winux::uint64 extendedPayloadLen;
    winux::uint8 maskingKey[4];

    FrameExtended64Masked()
    {
        memset( this, 0, sizeof(*this) );
    }

    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};

// 关闭帧载荷数据
struct CloseFramePayloadData
{
    winux::uint16 code;
    template < typename _Ty >
    _Ty * data()
    {
        return reinterpret_cast<_Ty*>( this + 1 );
    }
};


#pragma pack(pop)

// 发送WebSocket帧
EIENNET_FUNC_DECL(bool) SendWebSocketFrame( eiennet::Socket * sock, OpCode opcode, bool fin, bool mask, winux::byte * payloadData = nullptr, winux::uint payloadDataSize = 0 );
// 发送WebSocket数据
EIENNET_FUNC_DECL(bool) SendWebSocketBuffer( eiennet::Socket * sock, OpCode opcode, bool mask, winux::Buffer payloadData = winux::Buffer(), int perFrameMaxPayloadSize = -1 );
// 发送WebSocket数据
EIENNET_FUNC_DECL(bool) SendWebSocketAnsi( eiennet::Socket * sock, OpCode opcode, bool mask, winux::AnsiString payloadData = winux::AnsiString(), int perFrameMaxPayloadSize = -1 );



struct WebSocketWrapper
{
    WebSocketWrapper( ClientCtx * clientCtx ) : clientCtx(clientCtx)
    {
    }
    ClientCtx * clientCtx;
    winux::AnsiString messageBuf;  // websocket消息缓冲区，用于接收多帧数据（一条消息可能分片）
    int messageType = 0;   // 消息类型，1文本；2二进制
    ws::ReadyState state = ws::stateConnecting; // 连接状态

    bool close( winux::uint16 errCode = -1, winux::AnsiString const & errStr = "" )
    {
        winux::GrowBuffer data;
        if ( errCode != (winux::uint16)-1 )
        {
            errCode = htont(errCode);
            data.append( &errCode, sizeof(errCode) );
            data.append(errStr);
        }
        if ( !SendWebSocketBuffer( clientCtx->clientSockPtr.get(), ws::ctrlClose, false, std::move(data) ) ) return false;

        if ( this->state == ws::stateClosing )
            this->state = ws::stateClosed;
        else
            this->state = ws::stateClosing;
        return true;
    }

    bool send( winux::AnsiString const & data, ws::OpCode opcode = ws::dataText )
    {
        return SendWebSocketAnsi( clientCtx->clientSockPtr.get(), opcode, false, data, 65535 );
    }
};

class WsHttpClientCtx : public ClientCtx
{
public:
    WsHttpClientCtx( winux::uint64 clientId, winux::String clientEpStr, winux::SharedPointer<ip::tcp::Socket> clientSockPtr )
        : ClientCtx( clientId, clientEpStr, clientSockPtr ), url(http::Url::urlSimple), websocket(this)
    {
    }
    virtual ~WsHttpClientCtx()
    {
        ColorOutput( winux::fgPurple, "Client[", clientId, "]客户场景`", clientEpStr, "`析构" );
    }
    DataRecvSendCtx forClient; // 数据收发场景
    winux::AnsiString requestHeaderStr;   // 请求头字符串
    http::Header requestHeader;    // 请求头
    winux::GrowBuffer requestBody; // 请求体数据
    http::Url url; // URL

    bool isWebSocketWrapper = false;   // 是否启用websocket包装
    WebSocketWrapper websocket;
};

template < class _ClientCtx = WsHttpClientCtx >
class WsHttpServer : public Server<_ClientCtx>
{
public:
    using ClientCtxSharedPointer = typename Server<_ClientCtx>::ClientCtxSharedPointer;

    using OpenHandlerFunction = std::function< void( ClientCtxSharedPointer clientCtxPtr ) >;
    using MessageHandlerFunction = std::function< void( ClientCtxSharedPointer clientCtxPtr, winux::AnsiString const & data, int messageType ) >;
    using CloseHandlerFunction = std::function< void( ClientCtxSharedPointer clientCtxPtr, winux::uint16 errCode, winux::AnsiString const & errStr ) >;
    using ErrorHandlerFunction = std::function< void( ClientCtxSharedPointer clientCtxPtr, WebSocketErrorCode ec ) >;

    WsHttpServer( winux::String const & serverIp, winux::ushort port, int threadCount = 10, int listenBacklog = 10, double durationSec = 0.1 )
        : Server<_ClientCtx>( ip::EndPoint( serverIp, port ), threadCount, listenBacklog, durationSec )
    {
    }

    void onOpenHandler( OpenHandlerFunction handler ) { _openHandler = handler; }
    void onMessageHandler( MessageHandlerFunction handler ) { _messageHandler = handler; }
    void onCloseHandler( CloseHandlerFunction handler ) { _closeHandler = handler; }
    void onErrorHandler( ErrorHandlerFunction handler ) { _errorHandler = handler; }

protected:
    // 业务启动函数
    virtual void onStartup( ClientCtxSharedPointer clientCtxPtr ) override
    {
        ColorOutput( winux::fgYellow, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`到来" );
        this->_pool.task( &WsHttpServer::_doRecvRequestHeaderTask, this, clientCtxPtr );
    }

    // 尝试接收一个请求头的任务
    void _doRecvRequestHeaderTask( ClientCtxSharedPointer clientCtxPtr )
    {
        int rcWait;
        int rc = clientCtxPtr->clientSockPtr->recvWaitUntilTarget(
            "\r\n\r\n",
            &clientCtxPtr->forClient.data,
            &clientCtxPtr->forClient.extraData,
            &clientCtxPtr->forClient.hadBytes,
            &clientCtxPtr->forClient.startpos,
            &clientCtxPtr->forClient.pos,
            this->_pool.getDurationSec(),
            &rcWait,
            [&clientCtxPtr] ( int had, void* ) {
                clientCtxPtr->forClient.retryCount = 0; // 重置次数
            }
        );

        if ( clientCtxPtr->forClient.pos != -1 ) // 接收成功
        {
            // 解析请求头
            clientCtxPtr->requestHeaderStr = clientCtxPtr->forClient.data.toAnsi();
            clientCtxPtr->requestHeader.clear();
            clientCtxPtr->requestHeader.parse(clientCtxPtr->requestHeaderStr);

            clientCtxPtr->forClient.resetStatus(); // 重置收发状态
            clientCtxPtr->forClient.data = std::move(clientCtxPtr->forClient.extraData); // 把额外接收的数据当作请求体数据
            // 从Content-Length获取请求体大小
            int contentLength = static_cast<http::Header&>(clientCtxPtr->requestHeader).getHeader<int>("Content-Length");
            clientCtxPtr->forClient.targetBytes = contentLength - clientCtxPtr->forClient.data.getSize(); // 需要接收的目标大小
            if ( clientCtxPtr->forClient.targetBytes > 0 )
            {
                // 投递尝试接收请求体
                this->_pool.task( &WsHttpServer::_doRecvRequestBodyTask, this, clientCtxPtr );
            }
            else
            {
                // 投递处理请求任务
                this->_pool.task( &WsHttpServer::_doRequestTask, this, clientCtxPtr );
            }
        }
        else if ( rcWait == 0 )
        {
            clientCtxPtr->forClient.retryCount++;
            if ( clientCtxPtr->forClient.retryCount < DataRecvSendCtx::RetryCount )
            {
                // 重投尝试接收请求头
                this->_pool.task( &WsHttpServer::_doRecvRequestHeaderTask, this, clientCtxPtr );
            }
            else
            {
                ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]从客户`", clientCtxPtr->clientEpStr, "`接收一个请求头时长时间没有数据到来" );
                ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
                this->removeClient(clientCtxPtr->clientId);
            }
        }
        else
        {
            ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]从客户`", clientCtxPtr->clientEpStr, "`接收头部出错" );
            ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
            this->removeClient(clientCtxPtr->clientId);
        }
    }

    // 尝试接收一个请求体的任务
    void _doRecvRequestBodyTask( ClientCtxSharedPointer clientCtxPtr )
    {
        int rcWait;
        int rc = clientCtxPtr->clientSockPtr->recvWaitUntilSize(
            clientCtxPtr->forClient.targetBytes,
            &clientCtxPtr->forClient.data,
            &clientCtxPtr->forClient.hadBytes,
            this->_pool.getDurationSec(),
            &rcWait,
            [&clientCtxPtr] ( int had, void* ) {
                clientCtxPtr->forClient.retryCount = 0;
            }
        );

        if ( clientCtxPtr->forClient.hadBytes == clientCtxPtr->forClient.targetBytes ) // 接收完成
        {
            // 保存请求体数据
            clientCtxPtr->requestBody = std::move(clientCtxPtr->forClient.data);
            clientCtxPtr->forClient.resetStatus(); // 重置收发状态

            // 投递处理请求任务
            this->_pool.task( &WsHttpServer::_doRequestTask, this, clientCtxPtr );
        }
        else if ( rcWait == 0 )
        {
            clientCtxPtr->forClient.retryCount++;
            if ( clientCtxPtr->forClient.retryCount < DataRecvSendCtx::RetryCount )
            {
                // 重投尝试接收请求体
                this->_pool.task( &WsHttpServer::_doRecvRequestBodyTask, this, clientCtxPtr );
            }
            else
            {
                ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]从客户`", clientCtxPtr->clientEpStr, "`接收一个请求体时长时间没有数据到来" );
                ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
                this->removeClient(clientCtxPtr->clientId);
            }
        }
        else
        {
            ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]从客户`", clientCtxPtr->clientEpStr, "`接收请求体出错" );
            ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
            this->removeClient(clientCtxPtr->clientId);
        }
    }

    // 处理一个请求的任务
    void _doRequestTask( ClientCtxSharedPointer clientCtxPtr )
    {
        ColorOutput( winux::fgGreen, "Client[", clientCtxPtr->clientId, "]从客户`", clientCtxPtr->clientEpStr, "`接收请求头(bytes:", clientCtxPtr->requestHeaderStr.size(), ")" );
        std::cout << clientCtxPtr->requestHeaderStr;
        ColorOutput( winux::fgAtrovirens, "Client[", clientCtxPtr->clientId, "]从客户`", clientCtxPtr->clientEpStr, "`接收请求体(bytes:", clientCtxPtr->requestBody.getSize(), ")\n" );

        // 构造响应
        std::stringbuf rspBuf;
        std::ostream rspOut(&rspBuf);
        http::Header rspHdr;
        rspHdr.setResponseLine( "HTTP/1.1 200 OK", false );
        rspHdr["Content-Type"] = "text/html";

        clientCtxPtr->url.clear();
        clientCtxPtr->url.parse( clientCtxPtr->requestHeader.getUrl() );

        // 调用具体业务处理
        this->_webProcess( clientCtxPtr, clientCtxPtr->requestHeader, clientCtxPtr->url, rspHdr, rspOut );

        // 输出响应
        winux::AnsiString rspStr = rspBuf.str();
        if ( rspStr.size() > 0 )
            rspHdr("Content-Length") << rspStr.size();

        // 发送响应
        if ( clientCtxPtr->clientSockPtr->sendUntil( rspHdr.toString() + rspStr ) )
        {
            // 如果不保活就删除连接，否则继续投请求任务
            if ( winux::StrLower( clientCtxPtr->requestHeader["Connection"] ) == "keep-alive" )
            {
                // 尝试接收下一个请求头
                this->_pool.task( &WsHttpServer::_doRecvRequestHeaderTask, this, clientCtxPtr );
            }
            else if ( clientCtxPtr->isWebSocketWrapper ) // 是否启用websocket包装，投递websocket处理任务
            {
                // 已经成功建立websocket连接
                clientCtxPtr->websocket.state = ws::stateOpen;
                // 调用onopen事件虚函数
                this->onOpen(clientCtxPtr);

                // 投递接收websocket帧任务
                this->_pool.task( &WsHttpServer::_doRecvWebSocketFrameTask, this, clientCtxPtr );
            }
            else
            {
                ColorOutput( winux::fgAqua, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`不保活，通信完毕即关闭" );
                ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
                this->removeClient(clientCtxPtr->clientId);
            }
        }
        else // 发送响应失败
        {
            if ( clientCtxPtr->isWebSocketWrapper )
            {
                // websocket出错,连接意外失效
                this->onError( clientCtxPtr, errCnnUnexpectedInvalid );
            }
            ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`发送响应失败" );
            ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
            this->removeClient(clientCtxPtr->clientId);
        }
    }

    // 具体业务处理
    void _webProcess( ClientCtxSharedPointer & clientCtxPtr, http::Header const & reqHdr, http::Url const & url, http::Header & rspHdr, std::ostream & rspOut )
    {
        // 是否升级为websocket连接
        if (
            reqHdr.hasHeader("Upgrade") && reqHdr.getHeader("Upgrade") == "websocket" &&
            reqHdr.hasHeader("Connection") && reqHdr.getHeader("Connection") == "Upgrade" &&
            reqHdr.hasHeader("Sec-WebSocket-Key")
        )
        {
            rspHdr.clear();
            rspHdr.setResponseLine( "HTTP/1.1 101 Switching Protocols", false );
            rspHdr["Upgrade"] = "websocket";
            rspHdr["Connection"] = "Upgrade";
            winux::String secWebsocketAccept = Base64Encode( winux::Sha1( reqHdr.getHeader("Sec-WebSocket-Key") + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" ) );
            rspHdr["Sec-WebSocket-Accept"] = secWebsocketAccept;

            ColorOutput( winux::fgAqua, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`升级为websocket连接" );
            std::cout << rspHdr.toString();

            // 正在升级为websocket
            clientCtxPtr->websocket.state = ws::stateConnecting;

            // 启用WebSocket包装
            clientCtxPtr->isWebSocketWrapper = true;
        }
        else
        {
            http::Vars get( url.getRawQueryStr().c_str() ); // GET变量
            winux::String urlPathStr = url.getPath();
            if ( 1 ) //urlPathStr == "" )
            {
                rspOut << "This is a websocket server!";
            }
        }
    }

    // 接收一个WebSocket帧的任务
    void _doRecvWebSocketFrameTask( ClientCtxSharedPointer clientCtxPtr )
    {
        winux::GrowBuffer frameBuf;
        ws::FrameBase * frame0;
        winux::int64 wantBytes = 0;
        winux::uint8 frameType = 0;

        // 先读取2字节的帧基础
        if ( !clientCtxPtr->clientSockPtr->recvUntilSize( sizeof(ws::FrameBase), &frameBuf ) )
        {
            // websocket出错,连接意外失效
            this->onError( clientCtxPtr, errCnnUnexpectedInvalid );
            ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`读取帧基础失败" );
            goto RecvFrameFailure;
        }

        frame0 = frameBuf.getBuf<ws::FrameBase>();

        // 判断并读取载荷数据长度
        if ( frame0->payloadLen < 126 )
        {
            frameType |= ws::ftBasic;
            auto frame1 = frameBuf.getBuf<ws::FrameBasic>();
            wantBytes += frame1->payloadLen;
        }
        else if ( frame0->payloadLen == 126 )
        {
            frameType |= ws::ftExtended16;
            if ( !clientCtxPtr->clientSockPtr->recvUntilSize( sizeof(ws::FrameExtended16::extendedPayloadLen), &frameBuf ) )
            {
                // websocket出错,连接意外失效
                this->onError( clientCtxPtr, errCnnUnexpectedInvalid );
                ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`读取16位帧载荷长度失败" );
                goto RecvFrameFailure;
            }
            auto len = ntoht( frameBuf.getBuf<ws::FrameExtended16>()->extendedPayloadLen ); // `extendedPayloadLen` is network byte order
            wantBytes += len;
        }
        else // 127
        {
            frameType |= ws::ftExtended64;
            if ( !clientCtxPtr->clientSockPtr->recvUntilSize( sizeof(ws::FrameExtended64::extendedPayloadLen), &frameBuf ) )
            {
                // websocket出错,连接意外失效
                this->onError( clientCtxPtr, errCnnUnexpectedInvalid );
                ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`读取64位帧载荷长度失败" );
                goto RecvFrameFailure;
            }
            auto len = ntoht( frameBuf.getBuf<ws::FrameExtended64>()->extendedPayloadLen ); // `extendedPayloadLen` is network byte order
            wantBytes += len;
        }

        // 是否有掩码
        if ( frame0->mask )
        {
            frameType |= 1;
            // 读取掩码
            if ( !clientCtxPtr->clientSockPtr->recvUntilSize( 4, &frameBuf ) )
            {
                // websocket出错,连接意外失效
                this->onError( clientCtxPtr, errCnnUnexpectedInvalid );
                ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`读取帧掩码失败" );
                goto RecvFrameFailure;
            }
        }

        // 读取载荷数据
        if ( !clientCtxPtr->clientSockPtr->recvUntilSize( (int)wantBytes, &frameBuf ) )
        {
            // websocket出错,连接意外失效
            this->onError( clientCtxPtr, errCnnUnexpectedInvalid );
            ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`读取载荷数据失败" );
            goto RecvFrameFailure;
        }

        // 进入帧处理
        switch ( frameType )
        {
        case ws::ftBasic:
            {
                auto frame = frameBuf.getBuf<ws::FrameBasic>();
                // 得到数据
                char * data = frame->data<char>();
                size_t len = frame->payloadLen;

                // 投递处理WebSocket帧任务
                this->_pool.task( &WsHttpServer::_doProcessWebSocketFrameTask, this, clientCtxPtr, (bool)frame->fin, (winux::uint)frame->opcode, winux::Buffer( data, len ) );
            }
            break;
        case ws::ftBasicMasked:
            {
                auto frame = frameBuf.getBuf<ws::FrameBasicMasked>();
                // 得到数据
                char * data = frame->data<char>();
                int len = frame->payloadLen;

                // 掩码算法
                for ( int i = 0; i < len; ++i )
                    data[i] ^= frame->maskingKey[ i % 4 ];

                // 投递处理WebSocket帧任务
                this->_pool.task( &WsHttpServer::_doProcessWebSocketFrameTask, this, clientCtxPtr, (bool)frame->fin, (winux::uint)frame->opcode, winux::Buffer( data, len ) );
            }
            break;
        case ws::ftExtended16:
            {
                auto frame = frameBuf.getBuf<ws::FrameExtended16>();
                // 得到数据
                char * data = frame->data<char>();
                int len = ntoht(frame->extendedPayloadLen);

                // 投递处理WebSocket帧任务
                this->_pool.task( &WsHttpServer::_doProcessWebSocketFrameTask, this, clientCtxPtr, (bool)frame->fin, (winux::uint)frame->opcode, winux::Buffer( data, len ) );
            }
            break;
        case ws::ftExtended16Masked:
            {
                auto frame = frameBuf.getBuf<ws::FrameExtended16Masked>();
                // 得到数据
                char * data = frame->data<char>();
                int len = ntoht(frame->extendedPayloadLen);

                // 掩码算法
                for ( int i = 0; i < len; ++i )
                    data[i] ^= frame->maskingKey[ i % 4 ];

                // 投递处理WebSocket帧任务
                this->_pool.task( &WsHttpServer::_doProcessWebSocketFrameTask, this, clientCtxPtr, (bool)frame->fin, (winux::uint)frame->opcode, winux::Buffer( data, (winux::uint)len ) );
            }
            break;
        case ws::ftExtended64:
            {
                auto frame = frameBuf.getBuf<ws::FrameExtended64>();
                // 得到数据
                char * data = frame->data<char>();
                winux::int64 len = ntoht(frame->extendedPayloadLen);

                // 投递处理WebSocket帧任务
                this->_pool.task( &WsHttpServer::_doProcessWebSocketFrameTask, this, clientCtxPtr, (bool)frame->fin, (winux::uint)frame->opcode, winux::Buffer( data, (winux::uint)len ) );
            }
            break;
        case ws::ftExtended64Masked:
            {
                auto frame = frameBuf.getBuf<ws::FrameExtended64Masked>();
                // 得到数据
                char * data = frame->data<char>();
                winux::int64 len = ntoht(frame->extendedPayloadLen);

                // 掩码算法
                for ( winux::int64 i = 0; i < len; ++i )
                    data[i] ^= frame->maskingKey[ i % 4 ];

                // 投递处理WebSocket帧任务
                this->_pool.task( &WsHttpServer::_doProcessWebSocketFrameTask, this, clientCtxPtr, (bool)frame->fin, (winux::uint)frame->opcode, winux::Buffer( data, (winux::uint)len ) );
            }
            break;
        }

        return;

    RecvFrameFailure:
        ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
        this->removeClient(clientCtxPtr->clientId);
    }

    // 处理WebSocket帧的任务
    void _doProcessWebSocketFrameTask( ClientCtxSharedPointer clientCtxPtr, bool fin, winux::uint opcode, winux::Buffer & payloadData )
    {
        ColorOutput( winux::fgGreen, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`收到一个帧{fin:", (winux::uint)fin, ",opcode:", opcode, ",datasize:", payloadData.getSize(), "}" );

        switch ( opcode )
        {
        case ws::dataContinued:
            {
                if ( fin )
                {
                    clientCtxPtr->websocket.messageBuf.append( payloadData.getBuf<char>(), payloadData.getSize() );
                    this->onMessage( clientCtxPtr, clientCtxPtr->websocket.messageBuf, clientCtxPtr->websocket.messageType );
                    clientCtxPtr->websocket.messageBuf.clear();
                    clientCtxPtr->websocket.messageType = 0;
                }
                else
                {
                    clientCtxPtr->websocket.messageBuf.append( payloadData.getBuf<char>(), payloadData.getSize() );
                }

                // 投递接收下一个websocket帧的任务
                this->_pool.task( &WsHttpServer::_doRecvWebSocketFrameTask, this, clientCtxPtr );
            }
            break;
        case ws::dataText:
        case ws::dataBinary:
        case ws::dataRsv3:
        case ws::dataRsv4:
        case ws::dataRsv5:
        case ws::dataRsv6:
        case ws::dataRsv7:
            {
                if ( fin )
                {
                    clientCtxPtr->websocket.messageBuf.append( payloadData.getBuf<char>(), payloadData.getSize() );
                    clientCtxPtr->websocket.messageType = opcode;
                    this->onMessage( clientCtxPtr, clientCtxPtr->websocket.messageBuf, clientCtxPtr->websocket.messageType );
                    clientCtxPtr->websocket.messageBuf.clear();
                    clientCtxPtr->websocket.messageType = 0;
                }
                else
                {
                    clientCtxPtr->websocket.messageBuf.append( payloadData.getBuf<char>(), payloadData.getSize() );
                    clientCtxPtr->websocket.messageType = opcode;
                }

                // 投递接收下一个websocket帧的任务
                this->_pool.task( &WsHttpServer::_doRecvWebSocketFrameTask, this, clientCtxPtr );
            }
            break;
        case ws::ctrlClose:
            {
                // 收到一个close帧
                // 如果有载荷
                winux::uint16 errCode = 0;
                winux::AnsiString errStr;
                if ( payloadData.getSize() > 0 )
                {
                    auto closeData = payloadData.getBuf<ws::CloseFramePayloadData>();
                    errCode = ntoht(closeData->code);
                    errStr.assign( closeData->data<char>(), payloadData.getSize() - sizeof(ws::CloseFramePayloadData::code) );
                    ColorOutput( winux::fgMaroon, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`收到一个Close帧{errcode:", errCode, ",errstr:", errStr, "}" );
                }
                else // 没有载荷数据
                {
                    ColorOutput( winux::fgMaroon, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`收到一个Close帧" );
                }

                // 收到一个close帧，如果处在closing状态，应立即关闭连接并改为closed状态；如果不处在closing状态则改为closing，发送一个close帧到对方后改为closed。
                if ( clientCtxPtr->websocket.state == ws::stateClosing )
                {
                    clientCtxPtr->clientSockPtr->close();
                    clientCtxPtr->websocket.state = ws::stateClosed;

                    this->onClose( clientCtxPtr, errCode, errStr ); // 处理onclose事件

                    // 删除场景
                    ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
                    this->removeClient(clientCtxPtr->clientId);
                }
                else
                {
                    clientCtxPtr->websocket.state = ws::stateClosing;

                    if ( clientCtxPtr->websocket.close( errCode, errStr ) )
                    {
                        ColorOutput( winux::fgAqua, "服务器回了一个close帧" );
                        clientCtxPtr->clientSockPtr->close();
                        clientCtxPtr->websocket.state = ws::stateClosed;

                        this->onClose( clientCtxPtr, errCode, errStr ); // 处理onclose事件

                        ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
                        this->removeClient(clientCtxPtr->clientId);
                    }
                }
            }
            break;
        case ws::ctrlPing:
            {
                // 收到一个ping帧，回一个pong帧
                SendWebSocketAnsi( clientCtxPtr->clientSockPtr.get(), ws::ctrlPong, false );

                // 投递接收下一个websocket帧任务
                this->_pool.task( &WsHttpServer::_doRecvWebSocketFrameTask, this, clientCtxPtr );
            }
            break;
        case ws::ctrlPong:
            {
                // 收到pong帧，无视

                // 投递接收下一个websocket帧任务
                this->_pool.task( &WsHttpServer::_doRecvWebSocketFrameTask, this, clientCtxPtr );
            }
            break;
        default:
            // 意外帧
            ColorOutput( winux::fgRed, "Client[", clientCtxPtr->clientId, "]客户`", clientCtxPtr->clientEpStr, "`收到一个意外帧{fin:",(winux::uint)fin,",opcode:",opcode,",datasize:",payloadData.getSize(),"}" );
            ColorOutput( winux::fgFuchsia, "Client[", clientCtxPtr->clientId, "]移除客户`", clientCtxPtr->clientEpStr, "`，还剩", this->_clients.size() - 1, "客户" );
            this->removeClient(clientCtxPtr->clientId);
            break;
        }
    }

    virtual void onOpen( ClientCtxSharedPointer clientCtxPtr )
    {
        if ( _openHandler ) _openHandler(clientCtxPtr);
    }

    virtual void onMessage( ClientCtxSharedPointer clientCtxPtr, winux::AnsiString const & data, int messageType )
    {
        if ( _messageHandler ) _messageHandler( clientCtxPtr, data, messageType );
    }

    virtual void onClose( ClientCtxSharedPointer clientCtxPtr, winux::uint16 errCode, winux::AnsiString const & errStr )
    {
        if ( _closeHandler ) _closeHandler( clientCtxPtr, errCode, errStr );
    }

    virtual void onError( ClientCtxSharedPointer clientCtxPtr, WebSocketErrorCode ec )
    {
        if ( _errorHandler ) _errorHandler( clientCtxPtr, ec );
    }

    OpenHandlerFunction _openHandler;
    MessageHandlerFunction _messageHandler;
    CloseHandlerFunction _closeHandler;
    ErrorHandlerFunction _errorHandler;
};


} // namespace ws

} // namespace eiennet