WSHCommon/Common/CommonNetwork/Sockets/NetSessionAsync.cs

using System;
using System.Collections;
using System.Collections.Generic;
using System.Net.Sockets;
using System.IO;
using System.Threading;
using CommonNetwork.Net;
using CommonLang.IO;
using CommonLang.ByteOrder;
using CommonLang.Protocol;
using System.Net;
using CommonLang.Net;
using CommonLang;
using CommonLang.Log;

namespace CommonNetwork.Sockets
{
    public class NetSessionAsync : BaseNetSession
    {
        private Logger log = LoggerFactory.GetLogger("NetSessionAsync");

        private Socket mTCP = null;
        protected INetPackageCodec mCodec;
        private INetSessionListener mListener;

        private Queue<Object> mSendQueue = new Queue<Object>();

        public NetSessionAsync()
        {
        }

        /// <summary>
        /// 判断当前网络是否已经连接
        /// </summary>
        /// <returns></returns>
        public override bool IsConnected
        {
            get
            {
                Socket tcp = mTCP;
                if (tcp != null)
                {
                    return tcp.Connected;
                }
                return false;
            }
        }
        public override INetPackageCodec Codec
        {
            get { return mCodec; }
        }
        public override IPEndPoint RemoteAddress
        {
            get
            {
                if (mTCP != null) return mTCP.RemoteEndPoint as IPEndPoint;
                return null;
            }
        }
        public Socket Session
        {
            get { return mTCP; }
        }

        private void onException(Exception err)
        {
            mListener.onError(this, err);
            if (mOnError != null)
            {
                mOnError.Invoke(this, err);
            }
        }

        //-------------------------------------------------------------------------------------
        #region Open

        public override bool Open(string url, INetPackageCodec codec, INetSessionListener listener)
        {
            bool ret = false;
            try
            {
                lock (this)
                {
                    if (mTCP == null)
                    {
                        this.mCodec = codec;
                        this.mListener = listener;
                        this.mURL = url;
                        string[] url_kv = url.Split(':');
                        //this.mRemoteAddress = IPUtil.ToEndPoint(url_kv[0], int.Parse(url_kv[1]));
                        //new IPEndPoint(IPAddress.Parse(url_kv[0]), int.Parse(url_kv[1]));
                        lock (mSendQueue) this.mSendQueue.Clear();
                        // 建立SOCKET链接
                        this.mTCP = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                        //this.mTCP.ReceiveTimeout = 5000;
                        //this.mTCP.SendTimeout = 5000;
                        this.mTCP.NoDelay = true;
                        this.mTCP.BeginConnect(url_kv[0], int.Parse(url_kv[1]), endConnect, mTCP);
                        //创建读写线程对象
                        ret = true;
                    }
                }
            }
            catch (Exception err)
            {
                log.Error(err.Message, err);
                onException(new NetException("\n[Open:]" + URL +
                                                "\n[InnerException:]" + err.InnerException +
                                                "\n[Exception:]" + err.Message +
                                                "\n[Source:]" + err.Source +
                                                "\n[StackTrace:]" + err.StackTrace));
            }
            return ret;
        }

        private void endConnect(IAsyncResult result)
        {
            mTCP.EndConnect(result);
            if (result.IsCompleted)
            {
                onOpen();
                startReceiveHead();
            }
            else
            {
                Close();
            }
        }
        private void onOpen()
        {
            mListener.sessionOpened(this);
            if (mOnSessionOpened != null)
            {
                mOnSessionOpened.Invoke(this);
            }
        }

        #endregion
        //-------------------------------------------------------------------------------------
        #region Close

        public override bool Close()
        {
            bool ret = false;
            lock (this)
            {
                if (mTCP != null)
                {
                    try
                    {
                        this.mTCP.Close();
                    }
                    catch (Exception err)
                    {
                        log.Error(err.Message, err);
                    }
                    finally
                    {
                        lock (mSendQueue)
                        {
                            this.mSendQueue.Clear();
                        }
                    }
                    this.mTCP = null;
                    ret = true;
                    onClose();
                }
            }
            return ret;
        }

        private void onClose()
        {
            lock (mSendQueue)
            {
                mSendQueue.Clear();
            }
            mListener.sessionClosed(this);
            if (mOnSessionClosed != null)
            {
                mOnSessionClosed.Invoke(this);
            }
        }

        #endregion
        //-------------------------------------------------------------------------------------

        //-------------------------------------------------------------------------------------

        //-------------------------------------------------------------------------------------

        //-------------------------------------------------------------------------------------

        //-------------------------------------------------------------------------------------------------------------------------------
        #region Send

        /// <summary>
        /// 发送一个消息，该方法将立即返回。
        /// </summary>
        /// <param name="data"></param>
        public override void Send(Object data)
        {
            lock (this)
            {
                if (mTCP != null)
                {
                    lock (mSendQueue)
                    {
                        mSendQueue.Enqueue(data);
                    }
                    // 通知写线程开始工作。
                    startSend();
                }
            }
        }

        public override void SendResponse(IMessage rsponse, int requestMessageID)
        {
            rsponse.MessageID = requestMessageID;
            Send(rsponse);
        }

        private void onSent(Object message)
        {
            mListener.messageSent(this, message);
            if (mOnMessageSent != null)
            {
                mOnMessageSent.Invoke(this, message);
            }
        }

        private class SendObject
        {
            public readonly List<object> sending = new List<object>();
            public readonly MemoryStream buffer = new MemoryStream(1024);
        }
        private class SendObjectPool
        {
            private ObjectPool<SendObject> s_Pool = new ObjectPool<SendObject>(s_ListPool_OnCreate);
            private static SendObject s_ListPool_OnCreate()
            {
                return new SendObject();
            }
            public SendObject Alloc()
            {
                SendObject ret = s_Pool.Get();
                ret.buffer.Position = 0;
                ret.buffer.SetLength(0);
                ret.sending.Clear();
                return ret;
            }
            public void Release(SendObject toRelease)
            {
                toRelease.sending.Clear();
                s_Pool.Release(toRelease);
            }
        }
        private SendObjectPool mSendPool = new SendObjectPool();

        private void startSend()
        {
            var sending = mSendPool.Alloc();
            try
            {
                lock (mSendQueue)
                {
                    if (mSendQueue.Count > 0)
                    {
                        sending.sending.AddRange(mSendQueue);
                        mSendQueue.Clear();
                    }
                }
                if (sending.sending.Count > 0 && mTCP.Connected)
                {
                    for (int i = 0; i < sending.sending.Count; i++)
                    {
                        object send_msg = sending.sending[i];
                        doEncode(sending.buffer, send_msg);
                    }
                }
                mTCP.BeginSend(sending.buffer.GetBuffer(), 0, (int)sending.buffer.Length, SocketFlags.None, endSend, sending);
            }
            catch (Exception err)
            {
                mSendPool.Release(sending);
                log.Error(err.Message, err);
                onException(new NetException("\n[runWrite:]" + URL +
                            "\n[InnerException:]" + err.InnerException +
                            "\n[Exception:]" + err.Message +
                            "\n[Source:]" + err.Source +
                            "\n[StackTrace:]" + err.StackTrace, err));
                this.Close();
            }
        }
        private void endSend(IAsyncResult result)
        {
            if (result.IsCompleted)
            {
                var sending = result.AsyncState as SendObject;
                try
                {
                    int length = mTCP.EndSend(result);
                    if (length > 0)
                    {
                        mSendBytes += length;
                        mSendPacks += sending.sending.Count;
                        for (int i = 0; i < sending.sending.Count; i++)
                        {
                            object send_msg = sending.sending[i];
                            onSent(send_msg);
                        }
                        sending.sending.Clear();
                    }
                    else
                    {
                        Close();
                    }
                }
                catch (Exception err)
                {
                    log.Error(err.Message, err);
                    onException(new NetException("endReceive: " + err.Message));
                    this.Close();
                }
                finally
                {
                    mSendPool.Release(sending);
                }
            }
            else
            {
                log.Info("Continue send !");
            }
        }

        #endregion
        //-------------------------------------------------------------------------------------------------------------------------------
        #region Receive

        private void onReceive(Object message)
        {
            try
            {
                mListener.messageReceived(this, message);
                if (mOnMessageReceived != null)
                {
                    mOnMessageReceived.Invoke(this, message);
                }
            }
            catch (Exception err)
            {
                log.Error(err.Message, err);
                onException(err);
            }
        }
        private class ReceiveObject
        {
            public readonly byte[] head = new byte[4];
            public int head_position = 0;
            public int body_length = 0;
            public int body_position = 0;
            public readonly MemoryStream body_buffer = new MemoryStream(1024);
        }
        private class ReceiveObjectPool
        {
            private ObjectPool<ReceiveObject> s_Pool = new ObjectPool<ReceiveObject>(s_ListPool_OnCreate);
            private static ReceiveObject s_ListPool_OnCreate()
            {
                return new ReceiveObject();
            }
            public ReceiveObject Alloc()
            {
                ReceiveObject ret = s_Pool.Get();
                ret.body_buffer.Position = 0;
                ret.body_buffer.SetLength(0);
                ret.head_position = 0;
                ret.body_length = 0;
                ret.body_position = 0;
                return ret;
            }
            public void Release(ReceiveObject toRelease)
            {
                s_Pool.Release(toRelease);
            }
        }
        private ReceiveObjectPool mReceivePool = new ReceiveObjectPool();

        private void startReceiveHead()
        {
            var recv_object = mReceivePool.Alloc();
            try
            {
                mTCP.BeginReceive(
                    recv_object.head,
                    recv_object.head_position,
                    recv_object.head.Length - recv_object.head_position,
                    SocketFlags.None, endReceiveHead, recv_object);
            }
            catch (Exception err)
            {
                mReceivePool.Release(recv_object);
                log.Error(err.Message, err);
                onException(new NetException("endReceive: " + err.Message));
                this.Close();
            }
        }
        private void endReceiveHead(IAsyncResult result)
        {
            var recv_object = result.AsyncState as ReceiveObject;
            try
            {
                int length = mTCP.EndReceive(result);
                if (length > 0)
                {
                    mRecvBytes += length;
                    recv_object.head_position += length;
                    if (recv_object.head_position == recv_object.head.Length)
                    {
                        recv_object.body_length = GetBodyLength(recv_object.head);
                        recv_object.body_position = 0;
                        if (recv_object.body_buffer.Capacity < recv_object.body_length)
                        {
                            recv_object.body_buffer.Capacity = recv_object.body_length;
                        }
                        recv_object.body_buffer.SetLength(recv_object.body_length);
                        startReceiveBody(recv_object);
                    }
                    else if (recv_object.head_position > recv_object.head.Length)
                    {
                        throw new NetException("Receive head overfollow");
                    }
                    else
                    {
                        startReceiveHead();
                    }
                }
                else
                {
                    throw new NetException("Receive 0 bytes!");
                }
            }
            catch (Exception err)
            {
                mReceivePool.Release(recv_object);
                log.Error(err.Message, err);
                onException(new NetException("endReceive: " + err.Message));
                this.Close();
            }
        }
        private void startReceiveBody(ReceiveObject recv_object)
        {
            try
            {
                mTCP.BeginReceive(
                    recv_object.body_buffer.GetBuffer(),
                    recv_object.body_position,
                    recv_object.body_length - recv_object.body_position,
                    SocketFlags.None, endReceiveBody, recv_object);
            }
            catch (Exception err)
            {
                mReceivePool.Release(recv_object);
                log.Error(err.Message, err);
                onException(new NetException("endReceive: " + err.Message));
                this.Close();
            }
        }
        private void endReceiveBody(IAsyncResult result)
        {
            var recv_object = result.AsyncState as ReceiveObject;
            try
            {
                int length = mTCP.EndReceive(result);
                if (length > 0)
                {
                    mRecvBytes += length;
                    recv_object.body_position += length;
                    if (recv_object.body_position == recv_object.body_length)
                    {
                        Object msg = null;
                        var input = recv_object.body_buffer;
                        input.Position = 0;
                        if (doDecode(input, out msg))
                        {
                            mRecvPacks++;
                            onReceive(msg);
                        }
                        recv_object.head_position = 0;
                        startReceiveHead();
                        mReceivePool.Release(recv_object);
                    }
                    else if (recv_object.body_position > recv_object.body_length)
                    {
                        throw new NetException("Receive body overfollow");
                    }
                    else
                    {
                        startReceiveBody(recv_object);
                    }
                }
                else
                {
                    throw new NetException("Receive 0 bytes!");
                }
            }
            catch (Exception err)
            {
                mReceivePool.Release(recv_object);
                log.Error(err.Message, err);
                onException(new NetException("endReceive: " + err.Message));
                this.Close();
            }
        }
        #endregion
        //-------------------------------------------------------------------------------------------------------------------------------
        protected virtual int GetBodyLength(byte[] buffer)
        {
            int pos = 0;
            int length = LittleEdian.GetS32(buffer, ref pos);
            return length;
        }
        protected virtual bool doEncode(Stream output, object send_msg)
        {
            long old_position = output.Position;
            LittleEdian.PutS32(output, 0);
            if (Codec.doEncode(output, send_msg))
            {
                int full_length = (int)(output.Position - old_position);
                output.Position = old_position;
                LittleEdian.PutS32(output, full_length - 4);
                output.Position = old_position + full_length;
                return true;
            }
            return false;
        }
        protected virtual bool doDecode(Stream input, out object msg)
        {
            if (Codec.doDecode(input, out msg))
            {
                if (input.Position != input.Length)
                {
                    throw new Exception(string.Format("can not decode full trunk size={0} type={1}", input.Length, msg != null ? msg.GetType().FullName : msg));
                }
                return true;
            }
            return false;
        }

    }


}