BeetleX之WebSocket詳解

對於BeetleX來說編寫WebSocket服務是一件非常簡單的事情，當你實現一個Web Api應用的同時這些API方法也是WebSocket服務方法。接下來主要講解如何通過JavaScript調用BeetleX的WebSocket服務方法或定義一個適合自己數據格式的WebSocket服務。

引用組件

通過Nuget引用最新版本的BeetleX.FastHttpApi或通過下載源碼編譯組件

 

實現服務

由於組件支持基於類方法的方式來制定服務，所以定義一個服務非常簡單，以下是一個基於Websocket的hello world服務：

    [BeetleX.FastHttpApi.Controller]
    class Program
    {
        private static BeetleX.FastHttpApi.HttpApiServer mApiServer;

        static void Main(string[] args)
        {
            mApiServer = new BeetleX.FastHttpApi.HttpApiServer();
            mApiServer.Debug();
            mApiServer.Register(typeof(Program).Assembly);
            mApiServer.Open();
            Console.Write(mApiServer.BaseServer);
            Console.Read();
        }

        public string Hello(string name)
        {
            return $"{name} {DateTime.Now}";
        }
    }

JavaScript調用

由於組件定義一個調用規范，針對上面的方法調用有着一定的格式要求，大體的json格式如下:

{
     url:''
     params:{{"name":"value"},{"name1":"value1"}}
}

• url

  描述請求的方法路徑，針對以上示例對應的路徑是'/Hello',組件默認大小寫不敏感。

• params

  用於描述方法對應的參數列表

針對以上示例方法調用json如下:

{
      url: '/Hello', 
      params: { name: 'test' }
}

自己處理數據

組件的服務要求指定的請求格式和對應的響應格式，這樣對於一些使用者來說有些限制，如果不希望組件提供的格式而是自己制定數據方式的話可以綁定WebSocket數據接收事件，當事件綁定后組件會把接收的數據直接路由給事件來處理，不會再按原有的方式來解析處理。綁定事件如下:

  mApiServer.WebSocketReceive = (o, e) =>
            {
                Console.WriteLine(e.Frame.Body);
                var freame = e.CreateFrame($"{DateTime.Now}" + e.Frame.Body.ToString());
                e.Response(freame);
            };

不過這里的處理方式還是以文本為主，只是文本的格式解釋和輸出格式更多的進行控制。

處理非文本數據

默認情況都以文本的方式來處理數據，實際上websocket是支持二進制流的;如果希望在組件的基礎上自己處理二進制流數據需要制定一個數據解析器，解析器的接口規范如下：

    public interface IDataFrameSerializer
    {
        object FrameDeserialize(DataFrame data, PipeStream stream);//反序列化對象方法

        ArraySegment<byte> FrameSerialize(DataFrame packet, object body);//序列化方法

        void FrameRecovery(byte[] buffer);//Buffer回收方法

    }

組件默認的解析器實現如下:

        public virtual object FrameDeserialize(DataFrame data, PipeStream stream)
        {
            return stream.ReadString((int)data.Length);
        }

        private System.Collections.Concurrent.ConcurrentQueue<byte[]> mBuffers = new System.Collections.Concurrent.ConcurrentQueue<byte[]>();

        public virtual ArraySegment<byte> FrameSerialize(DataFrame data, object body)
        {
            byte[] result;
            if (!mBuffers.TryDequeue(out result))
            {
                result = new byte[this.Options.MaxBodyLength];
            }
            string value;
            if (body is string)
                value = (string)body;
            else
                value = Newtonsoft.Json.JsonConvert.SerializeObject(body);
            int length = Options.Encoding.GetBytes(value, 0, value.Length, result, 0);
            return new ArraySegment<byte>(result, 0, length);
        }

        public virtual void FrameRecovery(byte[] buffer)
        {
            mBuffers.Enqueue(buffer);
        }

在制定完成數據解析器后把它設置到FrameSerializer屬性上即可

HttpApiServer.FrameSerializer= new CustomFrameSerializer();

連接驗證

當通過瀏覽器訪問websocket服務的時候，在連接創建過程存在一個握手通訊包，這個通訊包一般都帶有用戶的Cookie,通過這個Cookie即可以驗證連接的來源，從而確定連接的有效性。組件提供一個WebSocketConnect事件來擴展這個驗證機制，事件制定如下:

            mApiServer.WebSocketConnect = (o, e) => {
                //e.Request.Header
                //e.Request.Cookies
                e.Cancel = true;
            };

使用者可以根據實際情況的需要判斷對應的數據來確定是否取消當前WebSocket連接。

基於流解釋WebSocket協議

網上有很多WebSocket協議解釋代碼，之前也寫過一份，不過都是針對byte[]進行分析，以下代碼是基於Stream的方式來分析協議，通過stream操作起來會更簡潔易懂

        internal DataPacketLoadStep Read(PipeStream stream)
        {
            if (mLoadStep == DataPacketLoadStep.None)
            {
                if (stream.Length >= 2)
                {
                    byte value = (byte)stream.ReadByte();
                    this.FIN = (value & CHECK_B8) > 0;
                    this.RSV1 = (value & CHECK_B7) > 0;
                    this.RSV2 = (value & CHECK_B6) > 0;
                    this.RSV3 = (value & CHECK_B5) > 0;
                    this.Type = (DataPacketType)(byte)(value & 0xF);
                    value = (byte)stream.ReadByte();
                    this.IsMask = (value & CHECK_B8) > 0;
                    this.PayloadLen = (byte)(value & 0x7F);
                    mLoadStep = DataPacketLoadStep.Header;
                }
            }
            if (mLoadStep == DataPacketLoadStep.Header)
            {
                if (this.PayloadLen == 127)
                {
                    if (stream.Length >= 8)
                    {
                        Length = stream.ReadUInt64();
                        mLoadStep = DataPacketLoadStep.Length;
                    }
                }
                else if (this.PayloadLen == 126)
                {
                    if (stream.Length >= 2)
                    {
                        Length = stream.ReadUInt16();
                        mLoadStep = DataPacketLoadStep.Length;
                    }
                }
                else
                {
                    this.Length = this.PayloadLen;
                    mLoadStep = DataPacketLoadStep.Length;
                }
            }
            if (mLoadStep == DataPacketLoadStep.Length)
            {
                if (IsMask)
                {
                    if (stream.Length >= 4)
                    {
                        this.MaskKey = new byte[4];
                        stream.Read(this.MaskKey, 0, 4);
                        mLoadStep = DataPacketLoadStep.Mask;
                    }
                }
                else
                {
                    mLoadStep = DataPacketLoadStep.Mask;
                }
            }
            if (mLoadStep == DataPacketLoadStep.Mask)
            {
                if (this.Length > 0 && (ulong)stream.Length >= this.Length)
                {
                    if (this.IsMask)
                        ReadMask(stream);
                    Body = this.DataPacketSerializer.FrameDeserialize(this, stream);
                    mLoadStep = DataPacketLoadStep.Completed;
                }
            }
            return mLoadStep;
        }

 

void IDataResponse.Write(PipeStream stream)
        {
            try
            {
                byte[] header = new byte[2];
                if (FIN)
                    header[0] |= CHECK_B8;
                if (RSV1)
                    header[0] |= CHECK_B7;
                if (RSV2)
                    header[0] |= CHECK_B6;
                if (RSV3)
                    header[0] |= CHECK_B5;
                header[0] |= (byte)Type;
                if (Body != null)
                {
                    ArraySegment<byte> data = this.DataPacketSerializer.FrameSerialize(this, Body);
                    try
                    {
                        if (MaskKey == null || MaskKey.Length != 4)
                            this.IsMask = false;
                        if (this.IsMask)
                        {
                            header[1] |= CHECK_B8;
                            int offset = data.Offset;
                            for (int i = offset; i < data.Count; i++)
                            {
                                data.Array[i] = (byte)(data.Array[i] ^ MaskKey[(i - offset) % 4]);
                            }
                        }
                        int len = data.Count;
                        if (len > 125 && len <= UInt16.MaxValue)
                        {
                            header[1] |= (byte)126;
                            stream.Write(header, 0, 2);
                            stream.Write((UInt16)len);
                        }
                        else if (len > UInt16.MaxValue)
                        {
                            header[1] |= (byte)127;
                            stream.Write(header, 0, 2);
                            stream.Write((ulong)len);
                        }
                        else
                        {
                            header[1] |= (byte)data.Count;
                            stream.Write(header, 0, 2);
                        }
                        if (IsMask)
                            stream.Write(MaskKey, 0, 4);
                        stream.Write(data.Array, data.Offset, data.Count);
                    }
                    finally
                    {
                        this.DataPacketSerializer.FrameRecovery(data.Array);
                    }
                }
                else
                {
                    stream.Write(header, 0, 2);
                }
            }
            finally
            {
                this.DataPacketSerializer = null;
                this.Body = null;
            }
        }

 

如果你對這代碼有興趣，最直接的方法是查看BeetleX的代碼https://github.com/IKende/FastHttpApi