C/S模型之消息傳輸

   前兩章分別講了Socket的數據傳輸,本章開始將在前兩章的基礎上構建一個消息傳輸機制.

   因為服務端和客戶端都離不開消息的傳輸,所以先講如何在前兩章的基礎上構建消息傳輸的信道,本例構造了一個類MessageSession,該類負責消息在網絡中傳送,並且該類實現了一個接口IMessageSession,其目的同前一章實現接口的目的一樣,都是為了方便以后的擴展,該類的主要傳輸主體為一個接口IMessage,可以說這是一個消息調用的接口,是為了遠程調用而定義的一個接口,本篇中先給出它的接口定義,其實現在后續章節中給出

IMessage接口的定義如下:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace YZKStdLibrary.Message
{
    /// <summary>
/// 消息接口
/// </summary>
    public interface IMessage
    {
        /// <summary>
/// 在服務進程中的消息唯一ID
/// </summary>
        long MessageID { get; }
        /// <summary>
/// 獲取登錄憑證,該憑證為經過服務端驗證的唯一標識
/// </summary>
        string LoginedAuthentication { get; }
        /// <summary>
/// 獲取調用成員名稱或Guid
/// </summary>
        string InvokeMemberGuid { get; }
        /// <summary>
/// 獲取處理該消息的服務類標識,使用該標識確定是哪個服務類處理該消息
/// </summary>
        string ServiceGuid { get; }
        /// <summary>
/// 獲取處理該消息的程序集
/// </summary>
        string AssemblyGuid { get; }
        /// <summary>
/// 消息參數數組
/// </summary>
        object[] Paramets { get; }
    }
}

同時還定義了一個傳輸進度的回調代理OnTransingCallback:

/// <summary>
/// 傳輸進度回調
/// </summary>
/// <param name="transesize">要傳輸的數據的字節數</param>
/// <param name="tranfferedbytes">已傳輸的字節數</param>
/// <param name="message">傳輸的消息</param>
/// <param name="transedirection">傳輸方向</param>
    public delegate void OnTransingCallback(long transesize, long tranfferedbytes, IMessage message, DataTranseDirection transedirection);

該類定義了三個異步傳輸方法,由於使用IAsyncResult的返回機制,所以沒有定義同不方法,要同步只需等待返回的完成信號即可,當然你也可以定義同步方法,這個在前一章的同步方法上構建比教簡單,有興趣的網友可以自己試試.三個異步傳輸方法分別是BeginTranseMessage,BeginSendMessage,BeginReceiveMessage,同時它們都對應一個結束方法EndTranseMessage,EndSendMessage,EndReceiveMessage,下面分別先說一下這三個方法的用途.

BeginTranseMessage:一般使用在客戶端操作,也就是使用在發送完數據並且等待響應數據的應用場景下,大家可以想一下,客戶端的調用一般都是一發一收的,即使是過程調用沒有返回數據你也應該返回一個調用結束的消息通知調用方是否成功調用或返回一個錯誤通知錯誤信息.

BeginSendMessage和BeginReceiveMessage;一般使用在服務端,也就是在先接收消息再返回消息響應的應用場景下.

下面講一下這三個方法:

BeginTranseMessageCore:先獲取參數ITranseParamets,並將它設置成發送的狀態,再設置傳輸時使用的流類型SR_StreamTypeEnum,然后開始數據傳輸,這里就涉及到接口ITranseParamets的實現類MessageTranseParamets

這個類在三個異步傳輸方法中都要用到,因此先講講這個類:MessageTranseParamets

SR_StreamType屬性:定義傳輸流的類型

WaitReceiveMessage屬性:定義是否等待消息響應

AsyncResult屬性:一個IAsyncResult接口的實現

p_Serialize變量:一個序列化接口的對象,該序列化接口是自定義的,當然也可以使用MS本身的序列化

p_BufferIndexe變量:這是一個使用的緩沖區索引的結構,該結構的目的是指明使用的緩沖區的開始索引和長度,由於本例是從一個更大的應用中摘錄的,在該應用中定義了一個緩沖區機制,即所有使用位數組的場景都使用同一個大的位數組,它是一開始就創建好的,使用者通過一些方法獲取它的可用開始索引和長度后使用,本例中的位數組流和傳輸緩沖區都是使用這樣一個機制.至於如何實現請網友們等待后續文章.

g_TempFileName變量:文件流的文件名稱

Message屬性:要發送的消息主體

TransingCallback屬性:傳輸進度回調代理

其它接口的屬性我就不講了,請參看上一章接口的定義.

TranseBuffer,TranseBufferOffset,TranseBufferSize大家先用new byte[]和0,byte[].Length代替

ResetToSendParamets方法:將參數設置成等待發送的狀態,該方法就是初始化發送時的參數,並且調用InitSendMessage方法初始化發送流,流的類型取決於屬性SR_StreamType,對於Auto默任使用位數組流,對於創建位數組流CreateBytesStream方法中的類型BytesBufferStream已說過在后續章節中給出,大家不妨先用CreateBytesStream2使用MemoryStream代替.並且將消息序列化到流中(此時可以根據約定進行加密或壓縮等操作),大家請注意了在InitSendMessage方法中最后設置了傳輸約定中的ITranseHeader.TranseSize和傳輸流的位置

ResetToRcvParamets方法:將參數設置成等待接收的狀態.

ChangeToRcvParamets方法:將參數送發送狀態轉為接收狀態

ResetToConnectParamets方法:將參數設置成連接狀態

HeaderTransed方法:該方法是接口中定義的方法,用於通訊模塊通知本接口約定已傳輸完成,在前一章中使用到本方法,

 在該方法中首先將約定已傳輸標識置位,並且如果是接收操作則根據約定初始化接收流,接收流的初始化也是根據SR_StreamType屬性創建的,所以發送和接收雙方都應該使用同一個SR_StreamType的值.同樣在沒有給出BytesBufferStream流的定義之前,大家先用MemoryStream代替

AsyncTranseOver方法:該方法是接口中定義的方法,用於通訊模塊通知本接口數據主體已傳輸完成,在前一章中使用到本方法,在該方法中如果是連接則直接調用連接完成的回調就完事,如果是接收則反序列化對象(此時可以根據約定進行解密或解壓等操作),最后完成傳輸(調用MessageSession.FinshSession方法)

AsyncNotyTransing:該方法是接口中定義的方法,用於通訊模塊通知本接口約定已傳輸完成,在前一章中使用到本方法,用於接收傳輸進度的通知,本例只是簡單調用OnTransingCallback回調(如果不為null的話)就完事,

以下為該類的定義:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using YZKStdLibrary.Network.Transe;
using System.Net;
using System.Threading;
using System.IO;
using System.IO.MemoryMappedFiles;
using YZKStdLibrary.IO;
using YZKStdLibrary.Serialize;

namespace YZKStdLibrary.Message.Channel
{
    internal class MessageTranseParamets : ITranseParamets
    {

        public MessageTranseParamets()
        {
            p_TranseHeader = new TranseHeader();
            p_AsyncResult = new MessageSessionAsyncResult(null, null);
            p_SR_StreamType = SR_StreamTypeEnum.Auto;
        }

        private SR_StreamTypeEnum p_SR_StreamType;

        public SR_StreamTypeEnum SR_StreamType
        {
            get { return p_SR_StreamType; }
            set { p_SR_StreamType = value; }
        }

        private bool p_WaitReceiveMessage;

        public bool WaitReceiveMessage
        {
            get { return p_WaitReceiveMessage; }
        }

        private MessageSession g_Owner;

        private MessageSessionAsyncResult p_AsyncResult;

        internal MessageSessionAsyncResult AsyncResult
        {
            get { return p_AsyncResult; }
        }

        private IFormattSerialize p_Serialize;

        private BufferIndexHandle p_BufferIndexe;

        private string g_TempFileName;

        private void DeleteTempFile()
        {
            if (!string.IsNullOrEmpty(g_TempFileName))
            {
                StandHelperMethod.DeleteFile(g_TempFileName);
                g_TempFileName = null;
            }
        }

        private void CloseStream()
        {
            if (p_TranseStream != null)
            {
                try
                {
                    p_TranseStream.Close();
                }
                catch { }
                try
                {
                    p_TranseStream.Dispose();
                }
                catch { }
                p_TranseStream = null;
            }
        }

        private void CreateMappFileStream()
        {
            g_TempFileName = MessageHelper.GetTempFileName();
            //MemoryMappedFile mp = MemoryMappedFile.CreateFromFile(g_TempFileName, FileMode.Create, null, 5 * 1024 * 1024);
//p_TranseStream = mp.CreateViewStream();
//mp.Dispose();
            p_TranseStream = new FileStream(g_TempFileName, FileMode.Create);
        }

        private void CreateBytesStream()
        {
            p_TranseStream = new BytesBufferStream(MessageHelper.StreamBuffer);
        }

        private void CreateBytesStream2()
        {
            p_TranseStream = new MemoryStream();
        }

        private IMessage p_Message;

        public IMessage Message
        {
            get { return p_Message; }
        }

        private OnTransingCallback p_TransingCallback;

        public OnTransingCallback TransingCallback
        {
            get { return p_TransingCallback; }
        }

        private bool TrySerializeMessage(IMessage message)
        {
            try
            {
                p_Serialize.Serialize(p_TranseStream, message);
            }
            catch (BytesBufferManagerAssignException)
            {
                return false;
            }
            catch (Exception e)
            {
                p_TranseError = e;
            }
            return true;
        }

        private bool InitSendMessage(IMessage message)
        {
            try
            {
                switch (p_SR_StreamType)
                {
                    case SR_StreamTypeEnum.Auto:
                        {
                            CreateBytesStream();
                            if (!TrySerializeMessage(message))
                            {
                                CloseStream();
                                CreateMappFileStream();
                                TrySerializeMessage(message);
                            }
                            break;
                        }
                    case SR_StreamTypeEnum.Bytes:
                        {
                            CreateBytesStream();
                            p_Serialize.Serialize(p_TranseStream, message);
                            break;
                        }
                    case SR_StreamTypeEnum.Files:
                        {
                            CreateMappFileStream();
                            p_Serialize.Serialize(p_TranseStream, message);
                            break;
                        }
                }
            }
            catch (Exception e)
            {
                p_TranseError = e;
            }
            
            if (p_TranseError != null)
                return false;
            p_Message = message;
            p_TranseHeader.TranseSize = p_TranseStream.Length;
            p_TranseStream.Seek(0, SeekOrigin.Begin);
            return true;
        }

        private void InitReceiveStream()
        {
            long x = p_TranseHeader.TranseSize;
            switch (p_SR_StreamType)
            {
                case SR_StreamTypeEnum.Auto:
                    {
                        if (x < MessageHelper.StreamBuffer.TotalBytesCount)
                        {
                            BufferIndexHandle bh;
                            if (x <= 0)
                                bh = MessageHelper.StreamBuffer.TryAssignBufferHandle();
                            else
                                bh = MessageHelper.StreamBuffer.TryAssignBufferHandle((int)x);
                            if (!bh.IsEmpty)
                            {
                                p_TranseStream = new BytesBufferStream(
                                    MessageHelper.StreamBuffer,
                                    bh,
                                    0, true, true, true, true);
                                //((BytesBufferStream)p_TranseStream).InitFixStream();
                            }
                            else
                                CreateMappFileStream();
                        }
                        else
                        {
                            CreateMappFileStream();
                        }
                        break;
                    }
                case SR_StreamTypeEnum.Bytes:
                    {
                        BufferIndexHandle bh;
                        if (x <= 0)
                            bh = MessageHelper.StreamBuffer.TryAssignBufferHandle();
                        else
                            bh = MessageHelper.StreamBuffer.TryAssignBufferHandle((int)x);
                        if (!bh.IsEmpty)
                        {
                            p_TranseStream = new BytesBufferStream(
                                MessageHelper.StreamBuffer,
                                bh,
                                0, true, true, true,true);
                            //((BytesBufferStream)p_TranseStream).InitFixStream();
                        }
                        break;
                    }
                case SR_StreamTypeEnum.Files:
                    {
                        CreateMappFileStream();
                        break;
                    }
            }
            
        }

        private void DeSerializeRcvMessage()
        {
            p_TranseStream.Seek(0, SeekOrigin.Begin);
            p_AsyncResult.Response =
                p_Serialize.DeSerialize(p_TranseStream) as IMessage;
        }

        #region ITranseParamets 成員

        private TranseHeader p_TranseHeader;

        public ITranseHeader TranseHeader
        {
            get { return p_TranseHeader; }
        }

        private Stream p_TranseStream;

        public Stream TranseStream
        {
            get { return p_TranseStream; }
        }

        public byte[] TranseBuffer
        {
            get { return MessageHelper.ReadWriteBuffer.Buffer; }
        }

        public int TranseBufferOffset
        {
            get { return p_BufferIndexe.Index; }
        }

        public int TranseBufferSize
        {
            get { return p_BufferIndexe.Count; }
        }

        private long p_BytesTransferred;

        public long BytesTransferred
        {
            get
            {
                return p_BytesTransferred;
            }
            set
            {
                p_BytesTransferred = value;
            }
        }

        private bool p_IsAsyncTranse;

        public bool IsAsyncTranse
        {
            get { return p_IsAsyncTranse; }
            set { p_IsAsyncTranse = value; }
        }

        private DataTranseDirection p_DataTranseDirection;

        public DataTranseDirection DataTranseDirection
        {
            get { return p_DataTranseDirection; }
        }

        private Exception p_TranseError;

        public Exception TranseError
        {
            get
            {
                return p_TranseError;
            }
            set
            {
                p_TranseError = value;
            }
        }

        private bool p_IsCancel;

        public bool IsCancel
        {
            get { return p_IsCancel; }
        }

        private bool p_HeaderTransedOver;

        public bool HeaderTransedOver
        {
            get { return p_HeaderTransedOver; }
        }

        private EndPoint p_ConnectToEndPoint;

        public EndPoint ConnectToEndPoint
        {
            get { return p_ConnectToEndPoint; }
        }

        private object p_ConnectTranseInterfaceSetting;

        public object ConnectTranseInterfaceSetting
        {
            get { return p_ConnectTranseInterfaceSetting; }
        }

        public void HeaderTransed()
        {
            p_HeaderTransedOver = true;
            if (p_DataTranseDirection == DataTranseDirection.ReceivingData)
            {
                InitReceiveStream();
            }
        }

        public void AsyncTranseOver()
        {
            if (p_DataTranseDirection == Network.Transe.DataTranseDirection.Connecting)
            {
                if (p_Cb != null)
                    p_Cb(this);
                return;
            }
            if (p_TranseError == null)
            {
                if (p_DataTranseDirection == Network.Transe.DataTranseDirection.ReceivingData)
                {
                    try
                    {
                        DeSerializeRcvMessage();
                    }
                    catch (Exception e)
                    {
                        p_TranseError = e;
                    }
                }
            }
            if (g_Owner != null)
                g_Owner.FinshSession(this);
        }

        public void AsyncNotyTransing()
        {
            if (p_TransingCallback != null)
                p_TransingCallback(p_TranseHeader.TranseSize, p_BytesTransferred, p_Message, p_DataTranseDirection);
        }

        #endregion

        internal void ResetToNull(bool asyncrest)
        {
            CloseStream();
            DeleteTempFile();
            MessageHelper.ReadWriteBuffer.RecoveryBufferHandle(p_BufferIndexe);
            p_BufferIndexe = BufferIndexHandle.Empty;
            if (asyncrest)
                p_AsyncResult.ResetToNull();
            else
                p_AsyncResult = null;
            p_Serialize = null;
            p_BytesTransferred = 0;
            p_DataTranseDirection = DataTranseDirection.Unkown;
            p_IsAsyncTranse = false;
            p_TranseError = null;
            p_IsCancel = false;
            p_HeaderTransedOver = false;
            p_ConnectToEndPoint = null;
            p_ConnectTranseInterfaceSetting = null;
            p_TransingCallback = null;
            p_Message = null;
            p_TranseHeader.RecyValue();
            g_Owner = null;
            p_WaitReceiveMessage = false;
            p_Cb = null;
        }

        internal bool ResetToSendParamets(
            MessageSession owner,
            IMessage sendmessage,
            int headerflag,
            object state,
            AsyncCallback callback,
            OnTransingCallback tccbk,
            bool waitrcvmsg)
        {
            g_Owner = owner;
            p_BufferIndexe = MessageHelper.ReadWriteBuffer.AssignBufferHandle();
            p_Serialize = CurrentAppConfiger.CurrentErpConfinger.Formattor;
            if (p_AsyncResult == null)
                p_AsyncResult = new MessageSessionAsyncResult(state, callback);
            else
                p_AsyncResult.Reset(state, callback);
            p_DataTranseDirection = Network.Transe.DataTranseDirection.SendingData;
            p_IsAsyncTranse = true;
            p_TransingCallback = tccbk;
            p_TranseHeader.TranseFlags = headerflag;
            p_WaitReceiveMessage = waitrcvmsg;
            return InitSendMessage(sendmessage);
        }

        internal void ResetToRcvParamets(
            MessageSession owner,
            int headerflag,
            object state,
            AsyncCallback callback,
            OnTransingCallback tccbk)
        {
            g_Owner = owner;
            p_BufferIndexe = MessageHelper.ReadWriteBuffer.AssignBufferHandle();
            p_Serialize = CurrentAppConfiger.CurrentErpConfinger.Formattor;
            if (p_AsyncResult == null)
                p_AsyncResult = new MessageSessionAsyncResult(state, callback);
            else
                p_AsyncResult.Reset(state, callback);
            p_DataTranseDirection = Network.Transe.DataTranseDirection.ReceivingData;
            p_IsAsyncTranse = true;
            p_TransingCallback = tccbk;
            p_TranseHeader.TranseFlags = headerflag;
            p_WaitReceiveMessage = false;
        }

        internal void ChangeToRcvParamets()
        {
            CloseStream();
            DeleteTempFile();
            p_Cb = null;
            p_BytesTransferred = 0;
            p_DataTranseDirection = DataTranseDirection.ReceivingData;
            p_IsAsyncTranse = true;
            p_TranseError = null;
            p_IsCancel = false;
            p_HeaderTransedOver = false;
            p_ConnectToEndPoint = null;
            p_ConnectTranseInterfaceSetting = null;
            p_Message = null;
            p_TranseHeader.TranseSize = 0;
            p_WaitReceiveMessage = false;
        }

        internal void ResetToConnectParamets(
            IMessage sendmessage,
            EndPoint ep,
            ConnectCallback cb,
            object facesetting,
            object state,
            OnTransingCallback tccbk)
        {
            p_TransingCallback = tccbk;
            p_IsAsyncTranse = true;
            p_ConnectToEndPoint = ep;
            p_DataTranseDirection = Network.Transe.DataTranseDirection.Connecting;
            p_ConnectTranseInterfaceSetting = facesetting;
            p_Cb = cb;
            p_Message = sendmessage;
            if (p_AsyncResult == null)
                p_AsyncResult = new MessageSessionAsyncResult(state, null);
            else
                p_AsyncResult.Reset(state, null);
        }

        private ConnectCallback p_Cb;
    }

    internal class MessageSessionAsyncResult : IAsyncResult,IDisposable
    {

        internal MessageSessionAsyncResult(object state,AsyncCallback callback)
        {
            p_AsyncState = state;
            p_CompletedSynchronously = false;
            p_IsCompleted = false;
            p_Callback = callback;
            p_AsyncWaitHandle = null;
            g_WaitLock = new object();
        }

        private object p_AsyncState;

        private bool p_CompletedSynchronously;

        private bool p_IsCompleted;

        private AsyncCallback p_Callback;

        private ManualResetEvent p_AsyncWaitHandle;

        private object g_WaitLock;

        #region IAsyncResult 成員

        public object AsyncState
        {
            get { return p_AsyncState; }
        }

        public WaitHandle AsyncWaitHandle
        {
            get
            {
                lock (g_WaitLock)
                {
                    if (p_AsyncWaitHandle == null)
                        p_AsyncWaitHandle = new ManualResetEvent(false);
                    return p_AsyncWaitHandle;
                }
            }
        }

        public bool CompletedSynchronously
        {
            get { return p_CompletedSynchronously; }
        }

        public bool IsCompleted
        {
            get { return p_IsCompleted; }
        }

        #endregion

        internal bool SetHandle()
        {
            p_IsCompleted = true;
            bool set = false;
            lock (g_WaitLock)
            {
                if (p_AsyncWaitHandle != null)
                {
                    p_AsyncWaitHandle.Set();
                    set = true;
                }
            }
            if (!set)
            {
                if (p_Callback != null)
                    p_Callback(this);
            }
            return set;
        }

        #region IDisposable 成員

        public void Dispose()
        {
            lock (g_WaitLock)
            {
                if (p_AsyncWaitHandle != null)
                {
                    try
                    {
                        p_AsyncWaitHandle.Close();
                    }
                    catch { }
                    try
                    {
                        p_AsyncWaitHandle.Dispose();
                    }
                    catch { }
                    p_AsyncWaitHandle = null;
                }
            }
        }

        #endregion

        private IMessage p_Response;

        internal IMessage Response
        {
            get { return p_Response; }
            set { p_Response = value; }
        }

        internal void ResetToNull()
        {
            Dispose();
            p_AsyncState = null;
            p_CompletedSynchronously = false;
            p_IsCompleted = false;
            p_Callback = null;
            p_Response = null;
            p_Error = null;
            p_ConnecteFace = null;
            p_Session = null;
        }

        internal void Reset(object state, AsyncCallback callback)
        {
            p_AsyncState = state;
            p_Callback = callback;
            p_IsCompleted = false;
        }

        private Exception p_Error;

        internal Exception Error
        {
            get { return p_Error; }
            set { p_Error = value; }
        }

        private ITranseInterface p_ConnecteFace;

        internal ITranseInterface ConnectFace
        {
            get { return p_ConnecteFace; }
            set { p_ConnecteFace = value; }
        }

        private IMessageSession p_Session;

        internal IMessageSession Session
        {
            get { return p_Session; }
            set { p_Session = value; }
        }
    }

}

    public enum SR_StreamTypeEnum
    {
        /// <summary>
/// 自動選擇
/// </summary>
        Auto,
        /// <summary>
/// 位數組流
/// </summary>
        Bytes,
        /// <summary>
/// 文件流
/// </summary>
        Files
    }

    internal delegate void ConnectCallback(MessageTranseParamets par);

BeginSendMessage方法:它和BeginTranseMessage方法一樣,只不過對waitrcv傳進False,也就是說該方法不等響應一旦發送完成就調用回調,而BeginTranseMessage方法要等待響應接收完成后再調用回調.

BeginReceiveMessage方法:它將參數ITranseParamets設為接收狀態后就開始等待接收數據,接收完成后就調用回調.

由於存在BeginSendMessage和BeginReceiveMessage方法,使用方可以進行N次發送和接收,這都是根據具體應用而定的,所以提供了Close和WaitClose方法由使用方負責在適當的時機關閉信道.

   下面是IMessage接口定義,先貼出來:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace YZKStdLibrary.Message
{
    /// <summary>
/// 消息接口
/// </summary>
    public interface IMessage
    {
        /// <summary>
/// 在服務進程中的消息唯一ID
/// </summary>
        long MessageID { get; }
        /// <summary>
/// 獲取登錄憑證,該憑證為經過服務端驗證的唯一標識
/// </summary>
        string LoginedAuthentication { get; }
        /// <summary>
/// 獲取調用成員名稱或Guid
/// </summary>
        string InvokeMemberGuid { get; }
        /// <summary>
/// 獲取處理該消息的服務類標識,使用該標識確定是哪個服務類處理該消息
/// </summary>
        string ServiceGuid { get; }
        /// <summary>
/// 獲取處理該消息的程序集
/// </summary>
        string AssemblyGuid { get; }
        /// <summary>
/// 消息參數數組
/// </summary>
        object[] Paramets { get; }
    }
}

 

  OK,這一章就先到這了.