miduo_client/Assets/LuaFramework/Scripts/Network/SocketClient.cs

949 lines
27 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

using UnityEngine;
using System;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Collections;
using System.Collections.Generic;
using GameLogic;
using System.Text;
using LuaInterface;
//using System.Runtime.InteropServices;
public enum NetworkStateType
{
Connected,
ConnectFail,
Reconnected,
ReconnectFail,
Exception,
Disconnect,
}
static class IPAddressExtensions
{
public static IPAddress MapToIPv6(this IPAddress addr)
{
if (addr.AddressFamily != AddressFamily.InterNetwork)
throw new ArgumentException("Must pass an IPv4 address to MapToIPv6");
string ipv4str = addr.ToString();
return IPAddress.Parse("::ffff:" + ipv4str);
}
public static bool IsIPv4MappedToIPv6(this IPAddress addr)
{
bool pass1 = addr.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6, pass2;
try
{
pass2 = (addr.ToString().StartsWith("0000:0000:0000:0000:0000:ffff:") ||
addr.ToString().StartsWith("0:0:0:0:0:ffff:") ||
addr.ToString().StartsWith("::ffff:")) &&
IPAddress.Parse(addr.ToString().Substring(addr.ToString().LastIndexOf(":") + 1)).AddressFamily == AddressFamily.InterNetwork;
}
catch
{
return false;
}
return pass1 && pass2;
}
}
public class SocketClient
{
public NetworkManager netMgr;
private TcpClient client = null;
private NetworkStream outStream = null;
private MemoryStream _memStream;
public MemoryStream CurMemoryStream
{
get
{
if (_memStream == null)
_memStream = new MemoryStream();
return _memStream;
}
}
private BinaryReader _reader;
public BinaryReader CurReader
{
get
{
if (_reader == null)
_reader = new BinaryReader(_memStream);
return _reader;
}
}
static readonly int HeaderSize = sizeof(ushort) + sizeof(int);
private const int MAX_READ = 8192;
private byte[] byteBuffer = new byte[MAX_READ];
static TEA crypto = new TEA();
Queue<NetMsg> mEvents = new Queue<NetMsg>();
Stack<NetworkStateInfo> stateInfoStack = new Stack<NetworkStateInfo>();
Dictionary<int, IDispatcher> dispatchers = new Dictionary<int, IDispatcher>();
Dictionary<int, IDispatcher> callbacks = new Dictionary<int, IDispatcher>();
int m_serialId;
int m_indicationSid;
bool m_isConnected = false;
bool m_isConnecting = false;
private const int INDICATION_RESPONSE = 10046;
private const int INDICATION_ERROR_CODE = 10400;
public string IpAddress { get; private set; }
public int Port { get; private set; }
// Use this for initialization
public SocketClient(string ipAddress, int port)
{
//XDebug.Log.l("****Socket*********************创建一个Socket ---" + ipAddress);
IpAddress = ipAddress;
Port = port;
_memStream = new MemoryStream();
_reader = new BinaryReader(_memStream);
}
public void SetIpAddress(string ipAddress, int port)
{
if (!IsConnected())
{
IpAddress = ipAddress;
Port = port;
m_indicationSid = 0;
m_serialId = 1;
}
}
/// <summary>
/// 连接服务器
/// </summary>
public void Connect()
{
//XDebug.Log.l("****Socket*********************Connect ---" + IpAddress);
Close();
try
{
var ipType = GetIPAddressType(IpAddress);
client = null;
client = new TcpClient(ipType);
client.SendTimeout = 1000;
client.ReceiveTimeout = 1000;
client.NoDelay = true;
client.BeginConnect(IpAddress, Port, new AsyncCallback(OnConnect), null);
}
catch (Exception e)
{
OnConnectFail();
Debug.LogWarning(e.Message);
}
}
public void OnConnectFail()
{
//XDebug.Log.l("****Socket*********************OnConnectFail ---" + IpAddress);
AddStateInfo(NetworkStateType.ConnectFail, null);
}
/// <summary>
/// 连接上服务器
/// </summary>
void OnConnect(IAsyncResult asr)
{
//XDebug.Log.l("****Socket*********************OnConnect ---" + IpAddress);
//outStream = client.GetStream();
//client.GetStream().BeginRead(byteBuffer, 0, MAX_READ, new AsyncCallback(OnRead), null);
//AddStateInfo(NetworkStateType.Connected, null);
m_isConnecting = false;
//检测是否成功的建立了链接。..
if (client == null || !client.Connected)
{
XDebug.Log.l("****Socket*********************OnConnect FAILED ---" + IpAddress);
return;
}
try
{
//完成本次链接
client.EndConnect(asr);
// 记录链接的成功事件。
outStream = client.GetStream();
client.GetStream().BeginRead(byteBuffer, 0, MAX_READ, new AsyncCallback(OnRead), null);
AddStateInfo(NetworkStateType.Connected, null);
XDebug.Log.l("****Socket*********************OnConnect SUCCESS ---" + IpAddress);
}
catch (Exception e)
{
XDebug.Log.l("****Socket*********************OnConnect FAILED ---" + e.Message);
}
}
/// <summary>
/// 写数据
/// </summary>
void WriteMessage(byte[] message)
{
MemoryStream ms = null;
using (ms = new MemoryStream())
{
ms.Position = 0;
BinaryWriter writer = new BinaryWriter(ms);
short msglen = (short)message.Length;
msglen = IPAddress.HostToNetworkOrder(msglen);
byte[] tempBuffer = new byte[message.Length];
Buffer.BlockCopy(message, 0, tempBuffer, 0, message.Length);
if (crypto != null)
tempBuffer = crypto.Encode(tempBuffer);
writer.Write(tempBuffer);
writer.Flush();
if (client != null && client.Connected)
{
//NetworkStream stream = client.GetStream();
byte[] payload = ms.ToArray();
outStream.BeginWrite(payload, 0, payload.Length, new AsyncCallback(OnWrite), ms);
}
else
{
XDebug.Log.l("client.connected----->>false");
}
}
}
int _NNN;
/// <summary>
/// 读取消息
/// </summary>
void OnRead(IAsyncResult asr)
{
int bytesRead = 0;
try
{
//System.Net.Sockets.Socket.SocketAsyncResult
//NetBug.Fix by FHF..
//甄别socket当前是否正在连接中避免重连的间隔内收到服务端消息。.
if (client == null || !client.Connected)
return;
//int ilen = client.GetStream().EndRead(asr);
// XDebug.Log.l("[SocketClient] From Socket: EndRead.Len:" + ilen);
lock (client.GetStream())
{
//读取字节流到缓冲区
bytesRead = client.GetStream().EndRead(asr);
}
if (bytesRead < 1)
{
//包尺寸有问题,断线处理
AddStateInfo(NetworkStateType.Disconnect, "bytesRead < 1");
return;
}
//XDebug.Log.l("[SocketClient] NowClient:@" + ++_NNN+" _:"+bytesRead + ":" + client.Client.Handle);
OnReceive(byteBuffer, bytesRead); //分析数据包内容,抛给逻辑层
lock (client.GetStream())
{
//分析完,再次监听服务器发过来的新消息
Array.Clear(byteBuffer, 0, byteBuffer.Length); //清空数组
client.GetStream().BeginRead(byteBuffer, 0, MAX_READ, new AsyncCallback(OnRead), null);
}
//XDebug.Log.l("SocketClient.OnRead " + byteBuffer.ToString());
}
catch (Exception ex)
{
//Debug.LogError("SocketClient.OnRead " + ex.ToString());
//Lnhbuc_PrintBytes();
AddStateInfo(NetworkStateType.Exception, ex.Message);
}
}
/// <summary>
/// 丢失链接
/// </summary>
public void AddStateInfo(NetworkStateType dis, string msg)
{
NetworkStateInfo info = new NetworkStateInfo();
info.type = dis;
info.msg = msg;
stateInfoStack.Push(info);
}
/// <summary>
/// 打印字节
/// </summary>
/// <param name="bytes"></param>
void Lnhbuc_PrintBytes()
{
string returnStr = string.Empty;
for (int i = 0; i < byteBuffer.Length; i++)
{
returnStr += byteBuffer[i].ToString("X2");
}
Debug.LogError(returnStr);
}
/// <summary>
/// 向链接写入数据流
/// </summary>
void OnWrite(IAsyncResult r)
{
try
{
MemoryStream stream = (MemoryStream)r.AsyncState;
outStream.EndWrite(r);
if (stream != null)
{
stream.Close();
}
}
catch (Exception ex)
{
Debug.LogError("OnWrite--->>>" + ex.Message);
}
}
/// <summary>
/// 接收到消息
/// </summary>
void OnReceive(byte[] bytes, int length)
{
//XDebug.Log.l("****Socket*********************OnReceive ---" + IpAddress);
try
{
var stream = CurMemoryStream;
var reader = CurReader;
stream.Seek(0, SeekOrigin.End);
stream.Write(bytes, 0, length);
//Reset to beginning
stream.Seek(0, SeekOrigin.Begin);
while (Bkupkw_RemainingBytes() > 4)
{
short messageLen = reader.ReadInt16();
messageLen = IPAddress.NetworkToHostOrder(messageLen);
//memStream.Seek(0, SeekOrigin.Begin);
//TEA解密需要补齐8位
int decodeMessageLen = (messageLen + 7) / 8 * 8;
if (Bkupkw_RemainingBytes() >= decodeMessageLen) //粘包处理
{
var msgBytes = reader.ReadBytes(decodeMessageLen);
if (crypto != null)
msgBytes = crypto.Decode(msgBytes, messageLen);
OnReceivedMessage(new ByteBuffer(msgBytes));
}
else
{
//Back up the position two bytes
stream.Position = stream.Position - 2;
break;
}
}
//Create a new stream with any leftover bytes
byte[] leftover = reader.ReadBytes((int)Bkupkw_RemainingBytes());
stream.SetLength(0); //Clear
stream.Write(leftover, 0, leftover.Length);
}
catch (Exception e)
{
Debug.LogError("Network Read Exception: " + e.ToString());
AddStateInfo(NetworkStateType.Disconnect, "Network Read Exception: " + e.ToString());
}
}
/// <summary>
/// 剩余的字节
/// </summary>
private long Bkupkw_RemainingBytes()
{
return CurMemoryStream.Length - CurMemoryStream.Position;
}
/// <summary>
/// 接收到消息
/// </summary>
/// <param name="ms"></param>
void OnReceivedMessage(ByteBuffer buffer)
{
//NetBug.Fix by FHF..
//甄别socket当前是否正在连接中避免重连的间隔内收到服务端消息。.
if (client == null || !client.Connected)
return;
//user id
buffer.ReadIntToByte();
//token
buffer.ReadIntToByte();
//协议号
int msgid = buffer.ReadIntToByte();
//XDebug.Log.l("[SocketClient:OnReceivedMessage]msgid:" + msgid);
//收到心跳包
if (msgid == 1001)
{
lock (mEvents)
{
mEvents.Enqueue(new NetMsg(msgid, 0, 0, buffer));
}
return;
}
//消息index
int sId = buffer.ReadIntToByte();
//结果码
int result = buffer.ReadIntToByte();
lock (mEvents)
{
mEvents.Enqueue(new NetMsg(msgid, sId, result, buffer));
}
}
private void Ygqmgx_indicationErrorAction(ByteBuffer buffer)
{
//XDebug.Log.l("****Socket*********************indicationErrorAction ---" + IpAddress);
Util.CallMethod("SocketManager", "ReceiveErrorInfo", this, buffer);
}
/// <summary>
/// 关闭链接
/// </summary>
public void Close()
{
//确保 关闭Client的链接状态。
UnregistNetMessage(INDICATION_ERROR_CODE, Ygqmgx_indicationErrorAction);
if (client != null)
{
try
{
client.Client.Disconnect(false);
//XDebug.Log.l("****Socket*********************Disconnect() ---" + client.Client.Handle+ "@"+ client.Client.LocalEndPoint.ToString());
}
catch { }
XDebug.Log.l("****Socket*********************Close ---" + client.Client.Handle);
try
{
client.Close();
client = null;
}
catch
{
}
}
m_isConnected = false;
m_isConnecting = false;
dispatchers.Clear();
callbacks.Clear();
stateInfoStack.Clear();
if (_memStream != null)
{
_memStream.Dispose();
_memStream = null;
}
if (_reader != null)
{
_reader.Close();
_reader = null;
}
Array.Clear(byteBuffer, 0, byteBuffer.Length);
//lock (mEvents)
//{
mEvents.Clear();
//}
}
bool testConnect;
public void Update()
{
// FHF DEBUG
//TimeSpan ts1 = Process.GetCurrentProcess().TotalProcessorTime;
//System.Diagnostics.Stopwatch stw = new System.Diagnostics.Stopwatch();
//stw.Start();
if (stateInfoStack.Count > 0)
{
var info = stateInfoStack.Pop();
if (info.type == NetworkStateType.Connected)
{
m_isConnected = true;
RegistNetMessage(INDICATION_ERROR_CODE, Ygqmgx_indicationErrorAction);
Util.CallMethod("SocketManager", "OnConnect", this);
}
else if (info.type == NetworkStateType.ConnectFail)
{
Util.CallMethod("SocketManager", "OnConnectFail", this);
}
else if (info.type == NetworkStateType.Reconnected)
{
m_isConnected = true;
RegistNetMessage(INDICATION_ERROR_CODE, Ygqmgx_indicationErrorAction);
Util.CallMethod("SocketManager", "OnReconnect", this);
}
else if (info.type == NetworkStateType.ReconnectFail)
{
Util.CallMethod("SocketManager", "OnReconnectFail", this);
}
else if (info.type == NetworkStateType.Disconnect)
{
Disconnect(" NetworkStateType.Disconnect");
}
else if (info.type == NetworkStateType.Exception)
{
Disconnect("NetworkStateType.Exception");
}
return;
}
// int _CCC = 0;
lock (mEvents)
{
while (mEvents.Count > 0)
{
DispatchMessage(mEvents.Dequeue());
// _CCC++;
}
}
testConnect = Input.GetKeyDown(KeyCode.F1);
if (testConnect)
{
XDebug.Log.l("~~~~~~~~~~~~~测试掉线~~~~~~~~~~~~~~~");
}
if (m_isConnecting && !testConnect)
{
return;
}
if ((testConnect || !IsConnected()) && m_isConnected)
{
Disconnect("TEST");
}
//FHF Debug time preform
// stw.Stop();
// if(_CCC>0)
// UnityEngine.XDebug.Log.l("NET EVENT HANDLE: " + _CCC.ToString()+"@"+(stw.ElapsedTicks).ToString());
}
void DispatchMessage(NetMsg _event)
{
int msgId = _event.msgId;
ByteBuffer msg = _event.msg;
//Debug.LogError("msgId:" + msgId);
//Debug.LogError("m_indicationSid:" + m_indicationSid);
//Debug.LogError("sid:" + _event.sid);
//处理心跳包回调
if (msgId == 1001)
{
Util.CallMethod("SocketManager", "ReceiveClientHeartBeat", this, msg);
return;
}
//处理错误码
if (_event.result == 0)
{
m_serialId++;
Util.CallMethod("SocketManager", "ReceiveErrorInfo", this, msg);
if (callbacks.ContainsKey(msgId))
{
callbacks.Remove(msgId);
}
return;
}
bool isDeal = false;
//处理推送消息当序列号比本地游标大1处理推送信息否则不处理
if (dispatchers.ContainsKey(msgId))
{
if (m_indicationSid == _event.sid - 1)
{
m_indicationSid++;
if (!dispatchers[msgId].Process(msg))
{
UnityEngine.Debug.LogErrorFormat("Failed to process message. msgId : {0}", msgId);
}
}
if (_event.sid <= m_indicationSid)
{
//Debug.LogError("INDICATION_RESPONSE" + msgId);
m_SendMessage(INDICATION_RESPONSE, null, _event.sid);
}
isDeal = true;
}
//处理请求回应消息
if (callbacks.ContainsKey(msgId))
{
var call = callbacks[msgId];
callbacks.Remove(msgId);
m_serialId++;
if (!call.Process(msg))
{
UnityEngine.Debug.LogErrorFormat("Failed to process message. msgId : {0}", msgId);
}
isDeal = true;
}
// 没有处理的消息
if(!isDeal && msgId != 10047)
{
Debug.LogError("*** 后端推送了未请求的回调消息或者未注册的 indication 消息:" + msgId);
// 回复未监听的消息
if (m_indicationSid == _event.sid - 1)
{
m_indicationSid++;
}
if (_event.sid <= m_indicationSid)
{
m_SendMessage(INDICATION_RESPONSE, null, _event.sid);
}
}
}
public string Error;
public void Disconnect(string _err)
{
Error = _err;
XDebug.Log.l("****Socket*********************Disconnect [" + Error + "] ---" + IpAddress);
Close();
Util.CallMethod("SocketManager", "OnDisconnect", this, _err);
}
private Coroutine checkConnect_co;
IEnumerator CheckConnect_Co()
{
yield return new WaitForEndOfFrame();
XDebug.Log.warning("Start Connecting IP: " + IpAddress + " Port: " + Port);
Connect();
m_isConnecting = true;
yield return new WaitForSeconds(AppConst.ConnectTimeout);
if (!IsConnected())
{
OnConnectFail();
}
m_isConnecting = false;
checkConnect_co = null;
}
/// <summary>
/// 发送连接请求
/// </summary>
public void TryConnect()
{
//XDebug.Log.l("****Socket*********************TryConnect ---" + IpAddress);
if (IsConnected())
{
NetworkStateInfo info = new NetworkStateInfo();
info.type = NetworkStateType.Connected;
info.msg = null;
stateInfoStack.Push(info);
return;
}
if (m_isConnecting)
return;
if (checkConnect_co != null)
{
netMgr.StopCoroutine(checkConnect_co);
}
checkConnect_co = netMgr.StartCoroutine(CheckConnect_Co());
}
private Coroutine reconnect_co;
IEnumerator CheckReconnect_Co()
{
m_isConnecting = true;
if (IsConnected())
{
Close();
yield return new WaitForSeconds(1);
}
Reconnect();
yield return new WaitForSeconds(AppConst.ConnectTimeout);
if (!IsConnected())
{
OnReconnectFail();
Close();
}
m_isConnecting = false;
reconnect_co = null;
}
float _LastReconectime = 0;
public void TryReconnect()
{
if (IsConnected())
{
NetworkStateInfo info = new NetworkStateInfo();
info.type = NetworkStateType.Reconnected;
info.msg = null;
stateInfoStack.Push(info);
return;
}
if (Time.time - _LastReconectime < AppConst.ConnectTimeout )
return;
_LastReconectime = Time.time;
XDebug.Log.l("****Socket*********************TryReconnect ---" + IpAddress);
if (reconnect_co != null)
{
netMgr.StopCoroutine(reconnect_co);
}
reconnect_co = netMgr.StartCoroutine(CheckReconnect_Co());
}
public void Reconnect()
{
XDebug.Log.l("****Socket*********************Reconnect ---" + IpAddress);
Close();
try
{
var ipType = GetIPAddressType(IpAddress);
client = null;
client = new TcpClient(ipType);
XDebug.Log.l("****Socket*********************NEW ---" + client.Client.Handle);
//client.SendTimeout = 1000;
//client.ReceiveTimeout = 1000;
client.NoDelay = true;
client.BeginConnect(IpAddress, Port, new AsyncCallback(OnReconnect), null);
}
catch (Exception e)
{
Debug.LogError(e.Message);
OnReconnectFail();
}
}
public AddressFamily GetIPAddressType(string ip)
{
bool hasIpv4 = false;
bool hasIpv6 = false;
try
{
IPAddress[] address = Dns.GetHostAddresses(ip);
for (int i = 0; i < address.Length; i++)
{
//从IP地址列表中筛选出IPv4类型的IP地址
//AddressFamily.InterNetwork表示此IP为IPv4,
//AddressFamily.InterNetworkV6表示此地址为IPv6类型
if (address[i].AddressFamily == AddressFamily.InterNetwork)
{
hasIpv4 = true;
}
else if (address[i].AddressFamily == AddressFamily.InterNetworkV6)
{
hasIpv6 = true;
}
Util.Log("Network IPAddresssFamily " + address[i].AddressFamily.ToString() + " IP: " + address[i].ToString());
}
}
catch (Exception ex)
{
Debug.LogWarning("Network GetIPAddressType Exception: " + ex);
return AddressFamily.InterNetwork;
}
if (hasIpv4)
return AddressFamily.InterNetwork;
if (hasIpv6)
return AddressFamily.InterNetworkV6;
return AddressFamily.InterNetwork;
}
public void OnReconnectFail()
{
XDebug.Log.l("****Socket*********************OnReconnectFail ---" + IpAddress);
AddStateInfo(NetworkStateType.ReconnectFail, null);
}
void OnReconnect(IAsyncResult result)
{
_LastReconectime = 0;
m_isConnecting = false;
//检测是否成功的建立了链接。..
if (client == null || !client.Connected)
{
XDebug.Log.l("****Socket*********************OnReconnect FAILED ---" + IpAddress);
return;
}
try
{
//完成本次链接
client.EndConnect(result);
// 记录链接的成功事件。
outStream = client.GetStream();
client.GetStream().BeginRead(byteBuffer, 0, MAX_READ, new AsyncCallback(OnRead), null);
AddStateInfo(NetworkStateType.Reconnected, null);
_LastReconectime = 0;
XDebug.Log.l("****Socket*********************OnReconnect SUCCESS @"+ client.Client.LocalEndPoint.ToString());
}
catch (Exception e)
{
XDebug.Log.l("****Socket*********************OnReconnect FAILED ---" + e.Message);
}
}
/// <summary>
/// 发送消息
/// </summary>
public void SendMessage(int msgId, ByteBuffer msg)
{
m_serialId++;
m_SendMessage(msgId, msg, m_serialId);
}
void m_SendMessage(int msgId, ByteBuffer msg, int sn)
{
if (!IsConnected())
{
return;
}
ByteBuffer buffer = new ByteBuffer();
buffer.WriteIntToByte(AppConst.UserId);
buffer.WriteIntToByte(AppConst.Token);
buffer.WriteIntToByte(msgId);
buffer.WriteIntToByte(sn);
if (msg != null)
{
buffer.WriteBuffer(new LuaInterface.LuaByteBuffer(msg.ToBytes()));
msg.Close();
}
WriteMessage(buffer.ToBytes());
buffer.Close();
//XDebug.Log.l("[SocketClient.m_SendMessage]..."+ msgId);
}
/// <summary>
/// 发送心跳
/// </summary>
public void SendHeartBeat()
{
ByteBuffer buffer = new ByteBuffer();
buffer.WriteIntToByte(AppConst.UserId);
buffer.WriteIntToByte(AppConst.Token);
buffer.WriteIntToByte(1000);
WriteMessage(buffer.ToBytes());
buffer.Close();
}
public bool IsConnected()
{
return client != null && client.Connected;
}
public void RegistNetMessage(int msgId, Action<ByteBuffer> handle)
{
if (!dispatchers.ContainsKey(msgId))
dispatchers.Add(msgId, new SimpleDispatcher());
//XDebug.Log.l("++++++++++++++++RegistNetMessage================================", msgId, IpAddress, Port);
SimpleDispatcher find = (SimpleDispatcher)dispatchers[msgId];
find.processor += handle;
}
public void UnregistNetMessage(int msgId, Action<ByteBuffer> handle)
{
if (!dispatchers.ContainsKey(msgId))
return;
dispatchers.Remove(msgId);
XDebug.Log.l("-------------------UnregistNetMessage================================", msgId, IpAddress, Port);
//SimpleDispatcher find = (SimpleDispatcher)dispatchers[msgId];
//find.processor -= handle;
}
public void SendMessageWithCallBack(int msgId, int receiveMsgId, ByteBuffer message, LuaFunction callback)
{
if (!IsConnected())
{
UnityEngine.Debug.LogErrorFormat("Try to send message with invalid connection.");
AddStateInfo(NetworkStateType.Disconnect, "bytesRead < 1");
}
else
{
SimpleDispatcher cb = new SimpleDispatcher();
cb.processor += b => callback.Call(b);
if (!callbacks.ContainsKey(receiveMsgId))
{
callbacks.Add(receiveMsgId, cb);
}
m_SendMessage(msgId, message, m_serialId);
}
}
}