ClientPeer

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Net.Sockets;

public class ClientPeer {

    public Socket ClientSocket { get; set; }

    public ClientPeer() {
        this.ReceiveArgs = new SocketAsyncEventArgs();
        this.ReceiveArgs.UserToken = this;
        this.ReceiveArgs.SetBuffer(new byte[1024],0,1024);
        this.SendArgs = new SocketAsyncEventArgs();
        this.SendArgs.Completed += SendArgs_Completed;

    }
    #region 辅助方法
    public string GetIP() {
        return ClientSocket.RemoteEndPoint.ToString().Split(':')[0];
    }
    public string GetPort() {
        return ClientSocket.RemoteEndPoint.ToString().Split(':')[1];
    }
    #endregion

    #region 接受数据 

    public delegate void ReceiveCompleted(ClientPeer client,SocketMsg msg);

    //一条消息解析完成的回调
    public ReceiveCompleted receiveCompleted;

    //数据缓存，一旦接收到数据就存到缓存里
    private List<byte> dataCache = new List<byte>();

    //接受的异步套接字请求
    public SocketAsyncEventArgs ReceiveArgs { get; set; }

    //是否正在处理接收的数据
    private bool isReceiveProcess = false;

    private byte[] buffer = new byte[102400];
    public void StartReceiveTask() {
        Task.Run(()=> {
            BeginReceive();
        });
    }

    private void BeginReceive() {
        ClientSocket.BeginReceive(buffer, 0, buffer.Length, SocketFlags.None, ReceiveCallBack, ClientSocket);
    }



    private void ReceiveCallBack(IAsyncResult ar) {
        try {
            int len = ClientSocket.EndReceive(ar);
            if (len > 0) {
                StartReceive(buffer.Take(len).ToArray());
                BeginReceive();
            } else {
                sendDisconnect(this, "客户端主动断开");
            }

        } catch (Exception) {

            sendDisconnect(this, "客户端主动断开");
        }
    }

    /// <summary>
    /// client自身处理数据包
    /// </summary>
    /// <param name="packet"></param>
    public void StartReceive(byte[] packet) {
        try {
            dataCache.AddRange(packet);
            if (isReceiveProcess == false) {
                processReceive();
            }
        } catch (Exception e) {
            Console.WriteLine(TimerManager.GetNowTime() + "-" + e .ToString());
        }
    }

    /// <summary>
    /// 处理接收到的数据
    /// </summary>
    private void processReceive() {
        try {
            isReceiveProcess = true;
            byte[] data = EncodeTool.DecodePacket(ref dataCache);
            if (data == null || data.Length <= 0 ) {
                isReceiveProcess = false;
                return;
            }
            SocketMsg msg = EncodeTool.DecodeMsg(data);

            //回调给上层
            if (receiveCompleted != null) {
                receiveCompleted(this,msg);
            }
            processReceive();
        } catch (Exception e) {
            Console.WriteLine(TimerManager.GetNowTime()+"-"+e.ToString());
        }
    }

    #endregion


    #region 发送数据
    private Queue<byte[]> sendQueue = new Queue<byte[]>();

    private bool isSendProcess = false;

    //用来发送数据的异步socket操作
    private SocketAsyncEventArgs SendArgs;

    public void Send(int opCode,int subCode, object value) {
        try {
            string jsonStr = EncodeTool.EnJsonObj(value);
            SocketMsg msg = new SocketMsg(opCode, subCode, jsonStr);
            Console.WriteLine(EncodeTool.EnJsonObj(msg));
            byte[] data = EncodeTool.EncodeMsg(msg);
            byte[] packet = EncodeTool.EncodePacket(data);
            Send(packet);
        } catch (Exception e) {

            Console.WriteLine(TimerManager.GetNowTime()+"-"+e.ToString());
        }
    }
    public void Send(byte[] packet) {
        try {
            sendQueue.Enqueue(packet);
            if (!isSendProcess) {
                send();
            }
        } catch (Exception e) {

            Console.WriteLine(TimerManager.GetNowTime()+"-"+e.ToString());
        }
    }

    private void send() {
        try {
            isSendProcess = true;
            //如果数据的条数等于0 停止发送
            if (sendQueue.Count <= 0) {
                isSendProcess = false;
                return;
            }
            //取出一条数据
            byte[] packet = sendQueue.Dequeue();
            SendArgs.SetBuffer(packet,0,packet.Length);
            bool result = ClientSocket.SendAsync(SendArgs);
            if (result == false) {
                processSend();
            }

        } catch (Exception e) {

            Console.WriteLine(TimerManager.GetNowTime()+"-"+e.ToString());
        }
    }

    //当数据发送到连接的socket对象时自动调用的事件
    private void SendArgs_Completed(object sender,SocketAsyncEventArgs e) {
        //继续发送
        processSend();
    }

    //发送请求完成时调用  判断发送是否成功
    private void processSend() {
        try {
            if (SendArgs.SocketError != SocketError.Success) {
                //发送错误 客户端断开连接停止发送
                sendDisconnect(this,SendArgs.SocketError.ToString());
            } else {
                send();
            }
        } catch (Exception e) {
            Console.WriteLine(TimerManager.GetNowTime()+"-"+e.ToString());
        }
    }

    #endregion

    #region 断开连接

    //发送的时候 发现断开的回调
    public delegate void SendDisconnect(ClientPeer client, string reason);

    public SendDisconnect sendDisconnect;

    public void Disconnect() {
        try {
            //清空数据
            dataCache.Clear();
            isReceiveProcess = false;
            sendQueue.Clear();
            isSendProcess = false;
            if (ClientSocket == null) {
                return;
            }
            //ClientSocket.Shutdown(SocketShutdown.Both);
            ClientSocket.Close();
            ClientSocket = null;
        } catch (Exception e) {
            Console.WriteLine(TimerManager.GetNowTime()+"-"+e.ToString());
        }
    }

    #endregion
}