ylsj/Assets/Scripts/Core/Socket/Utils/SocketManager.cs

using UnityEngine;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using System.Net;
using System;
using System.Text;
using System.IO;
using pb = Google.Protobuf;

///// <summary>
///// 聊天部分使用的
///// </summary>
//[Obsolete("-g",true)]
//public class SocketManager
//{
//    private static SocketManager _instance;
//    public static SocketManager Instance
//    {
//        get
//        {
//            if (_instance == null)
//            {
//                _instance = new SocketManager();


//            }
//            return _instance;
//        }
//    }
//    private string _currIP;
//    private int _currPort;

//    private bool _isConnected = false;
//    public bool IsConnceted { get { return _isConnected; } }
//    private Socket clientSocket = null;
//    private Thread receiveThread = null;

//    private DataBuffer _databuffer = new DataBuffer();

//    byte[] _tmpReceiveBuff = new byte[4096];
//    private sSocketData _socketData = new sSocketData();

//    /// <summary>
//    /// 断开
//    /// </summary>
//    private void _close()
//    {
//        if (!_isConnected)
//            return;

//        _isConnected = false;

//        if (receiveThread != null)
//        {
//            receiveThread.Abort();
//            receiveThread = null;
//        }

//        if (clientSocket != null && clientSocket.Connected)
//        {
//            clientSocket.Close();
//            clientSocket = null;
//        }
//    }

//    private void _ReConnect()
//    {
//    }

//    /// <summary>
//    /// 连接
//    /// </summary>
//    private void _onConnet()
//    {
//        try
//        {
//            clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);//创建套接字
//            IPAddress ipAddress = IPAddress.Parse(_currIP);                                            //解析IP地址
//            IPEndPoint ipEndpoint = new IPEndPoint(ipAddress, _currPort);
//            IAsyncResult result = clientSocket.BeginConnect(ipEndpoint, new AsyncCallback(_onConnect_Sucess), clientSocket);//异步连接
//            bool success = result.AsyncWaitHandle.WaitOne(5000, true);
//            if (!success) //超时
//            {
//                _onConnect_Outtime();
//            }
//            clientSocket.SetKeepAlive(1000);
//        }
//        catch (System.Exception _e)
//        {
//            _onConnect_Fail();
//        }
//    }

//    private void _onConnect_Sucess(IAsyncResult iar)
//    {
//        try
//        {
//            Socket client = (Socket)iar.AsyncState;
//            client.EndConnect(iar);

//            receiveThread = new Thread(new ThreadStart(_onReceiveSocket));
//            receiveThread.IsBackground = true;
//            receiveThread.Start();
//            _isConnected = true;
//             LogHelper.Log("连接成功");
//        }
//        catch (Exception _e)
//        {
//            Close();
//        }
//    }

//    private void _onConnect_Outtime()
//    {
//        _close();
//    }

//    private void _onConnect_Fail()
//    {
//        _close();
//    }

//    /// <summary>
//    /// 发送消息结果回掉，可判断当前网络状态
//    /// </summary>
//    /// <param name="asyncSend"></param>
//    private void _onSendMsg(IAsyncResult asyncSend)
//    {
//        try
//        {
//            Socket client = (Socket)asyncSend.AsyncState;
//            client.EndSend(asyncSend);
//        }
//        catch (Exception e)
//        {
//             LogHelper.Log("send msg exception:" + e.StackTrace);
//        }
//    }

//    /// <summary>
//    /// 接受网络数据
//    /// </summary>
//    private void _onReceiveSocket()
//    {
//        while (true)
//        {
//            if (!clientSocket.Connected)
//            {
//                _isConnected = false;
//                _ReConnect();
//                break;
//            }
//            try
//            {
//                int receiveLength = clientSocket.Receive(_tmpReceiveBuff);
//                if (receiveLength > 0)
//                {
//                    _databuffer.AddBuffer(_tmpReceiveBuff, receiveLength);//将收到的数据添加到缓存器中
//                    while (_databuffer.GetData(out _socketData))//取出一条完整数据
//                    {
//                        MessageCenter.Instance._netMessageDataQueue.Enqueue(_socketData);
//                    }
//                }
//            }
//            catch (Exception e)
//            {
//                clientSocket.Disconnect(true);
//                clientSocket.Shutdown(SocketShutdown.Both);
//                clientSocket.Close();
//                break;
//            }
//        }
//    }




//    /// <summary>
//    /// ProtoBuf序列化
//    /// </summary>
//    /// <param name="data"></param>
//    /// <returns></returns>
//    public static byte[] ProtoBuf_Serializer(pb::IMessage data)
//    {
//        using (var ms = new MemoryStream())
//        using (var os = new pb::CodedOutputStream(ms))
//        {
//            data.WriteTo(os);
//            os.Flush();
//            ms.Seek(0, SeekOrigin.Begin);
//            return ms.ToArray();
//        }
//    }

//    /// <summary>
//    /// ProtoBuf反序列化
//    /// </summary>
//    /// <typeparam name="T"></typeparam>
//    /// <param name="_data"></param>
//    /// <returns></returns>
//    public static T ProtoBuf_Deserialize<T>(byte[] _data) where T : pb::IMessage<T>, new()
//    {
//        var _parser = new pb::MessageParser<T>(() => new T());
//        return _parser.ParseFrom(_data);
//    }



//    /// <summary>
//    /// 连接服务器
//    /// </summary>
//    /// <param name="_currIP"></param>
//    /// <param name="_currPort"></param>
//    public void Connect(string _currIP, int _currPort)
//    {
//        if (!IsConnceted)
//        {
//            this._currIP = _currIP;
//            this._currPort = _currPort;
//            _onConnet();
//        }
//        else if (clientSocket == null || !clientSocket.Connected)
//        {
//            _ReConnect();
//        }
//    }

//    /// <summary>
//    /// 发送消息基本方法
//    /// </summary>
//    /// <param name="_protocalType"></param>
//    /// <param name="_data"></param>
//    private void SendMsgBase(eProtocalCommand _protocalType, byte[] _data)
//    {
//        if (clientSocket == null || !clientSocket.Connected)
//        {
//            _ReConnect();
//            return;
//        }

//        //byte[] _msgdata = DataToBytes(_protocalType, _data);
//        var sd = sSocketData.FromBytes(_protocalType, _data);
//        try
//        {
//            clientSocket.BeginSend(sd.ToBytes(), 0, sd.PackLen, SocketFlags.None, new AsyncCallback(_onSendMsg), clientSocket);
//        }
//        catch (System.Exception e)
//        {
//             LogHelper.Log(e.Message);
//            clientSocket.Disconnect(true);
//            clientSocket.Shutdown(SocketShutdown.Both);
//            clientSocket.Close();

//        }

//    }

//    /// <summary>
//    /// 以二进制方式发送
//    /// </summary>
//    /// <param name="_protocalType"></param>
//    /// <param name="_byteStreamBuff"></param>
//    public void SendMsg(eProtocalCommand _protocalType, ByteStreamBuff _byteStreamBuff)
//    {
//        SendMsgBase(_protocalType, _byteStreamBuff.ToArray());
//    }

//    /// <summary>
//    /// 以ProtoBuf方式发送
//    /// </summary>
//    /// <param name="_protocalType"></param>
//    /// <param name="data"></param>
//    public void SendMsg(eProtocalCommand _protocalType, pb::IMessage data)
//    {
//        SendMsgBase(_protocalType, ProtoBuf_Serializer(data));
//    }

//    public void Close()
//    {
//        _close();
//    }

//}
/// <summary>
/// boss战部分使用的
/// </summary>
public class TasPBSocketManager
{

    private string _currIP;
    private int _currPort;

    private bool _isConnected = false;
    public bool IsConnceted { get { return _isConnected; } }
    private Socket clientSocket = null;
    private Thread receiveThread = null;
    private Task _receiveTask = null;

    private DataBuffer _databuffer = new DataBuffer();

    byte[] _tmpReceiveBuff = new byte[4096];
    private sSocketData _socketData = new sSocketData();

    public Action OnConnected;

    public Action OnDisconnected;

    /// <summary>
    /// 断开
    /// </summary>
    public void Close()
    {
        if (!_isConnected)
            return;

        _isConnected = false;

        if (receiveThread != null)
        {
            receiveThread.Abort();
            receiveThread = null;
        }

        if (clientSocket != null && clientSocket.Connected)
        {
            //clientSocket.Disconnect(true);
            clientSocket.Shutdown(SocketShutdown.Both);
            clientSocket.Close();
            clientSocket = null;
        }
    }

    private void _ReConnect()
    {
    }
    /// <summary>
    /// 连接服务器
    /// </summary>
    /// <param name="_currIP"></param>
    /// <param name="_currPort"></param>
    public void Connect(string _currIP, int _currPort)
    {
        if (!IsConnceted)
        {
            this._currIP = _currIP;
            this._currPort = _currPort;
            _onConnet();
        }
        else if (clientSocket == null || !clientSocket.Connected)
        {
            _ReConnect();
        }
    }
    /// <summary>
    /// 连接
    /// </summary>
    private async void _onConnet()
    {
        try
        {
            clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);//创建套接字
            IPAddress ipAddress = IPAddress.Parse(_currIP);                                            //解析IP地址
            IPEndPoint ipEndpoint = new IPEndPoint(ipAddress, _currPort);

            await clientSocket.ConnectAsync(ipEndpoint).WaitAsync(TimeSpan.FromSeconds(5));
            clientSocket.SetKeepAlive(60 * 1000);
            receiveThread = new Thread(new ThreadStart(_onReceiveSocket));
            receiveThread.IsBackground = true;
            receiveThread.Start();
            _isConnected = true;
            LogHelper.Log("连接成功");
            OnConnected?.Invoke();

        }
        catch (TimeoutException e)
        {
            LogHelper.LogError("连接超时");
            Close();
        }
        catch (Exception _e)
        {
            LogHelper.LogError("连接失败" + _e.Message);
            Close();
        }
    }


    /// <summary>
    /// 接受网络数据
    /// </summary>
    private async void _onReceiveSocket()
    {
        while (true)
        {
            if (!clientSocket.Connected)
            {
                _isConnected = false;
                _ReConnect();
                break;
            }
            try
            {
                var receiveLength = await clientSocket.ReceiveAsync(new ArraySegment<byte>(_tmpReceiveBuff), SocketFlags.None);  // 将接收到的信息存入到内存缓冲区，并返回其字节数组的长度    
                if (receiveLength > 0)
                {
                    _databuffer.AddBuffer(_tmpReceiveBuff, receiveLength);//将收到的数据添加到缓存器中
                    while (_databuffer.GetData(out _socketData))//取出一条完整数据
                    {
                        MessageCenter.Instance._netMessageDataQueue.Enqueue(_socketData);
                    }
                }
                else
                {                                                                          // 收到0字节数据视为对端连接已断开
                    OnDisconnected?.Invoke();
                    LogHelper.Log("发现对端断开连接.");
                    Close();
                }
            }
            catch (Exception e)
            {
                Close();
                break;
            }
        }
    }


    /// <summary>
    /// ProtoBuf反序列化
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="_data"></param>
    /// <returns></returns>
    public static T ProtoBuf_Deserialize<T>(byte[] _data) where T : pb::IMessage<T>, new()
    {
        var _parser = new pb::MessageParser<T>(() => new T());
        return _parser.ParseFrom(_data);
    }


    /// <summary>
    /// 发送消息基本方法
    /// </summary> 
    private async void SendMsgBase()
    {
        if (this.IsConnceted)
        {
            if (clientSocket == null || !clientSocket.Connected)
            {
                _ReConnect();
                return;
            }
            while (_msgQueue.TryDequeue(out var kv))
            {
                LogHelper.Log($"send msg in queue {kv.Key}");

                var msg = sSocketData.FromIMsg(kv.Key, kv.Value);
                try
                {
                    await clientSocket.SendAsync(new ArraySegment<byte>(msg.ToBytes()), SocketFlags.None);
                }
                catch (Exception e)
                {
                    LogHelper.Log(e.Message);
                    Close();
                }
            }
        }
    }

    public void Update()
    {
        SendMsgBase();
    }

    /// <summary>
    /// 以ProtoBuf方式发送
    /// </summary>
    /// <param name="_protocalType"></param>
    /// <param name="data"></param>
    public void SendMsg(eProtocalCommand _protocalType, pb::IMessage data)
    {
        _msgQueue.Enqueue(new KeyValuePair<eProtocalCommand, pb.IMessage>(_protocalType, data));
    }

    private ConcurrentQueue<KeyValuePair<eProtocalCommand, pb::IMessage>> _msgQueue = new ConcurrentQueue<KeyValuePair<eProtocalCommand, pb.IMessage>>();


}