StandardSence/StandardScene.Core/TCP/AsyncTcpClient.cs

using SimpleCore.Library;
using System;
using System.Diagnostics;
using System.Globalization;
using System.Net;
using System.Net.NetworkInformation;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace StandardScene.TCP
{
    /// <summary>
    /// 异步 TCP 客户端
    /// </summary>
    public class AsyncTcpClient : IDisposable
    {
        private sealed class DatagramReadState
        {
            public TcpClient Client { get; set; }
            public byte[] Buffer { get; set; }
        }

        public event EventHandler<TcpDatagramReceivedEventArgs<byte[]>> DatagramReceived;       // 接收到数据报文事件
        public event EventHandler<TcpDatagramReceivedEventArgs<byte[]>> PlaintextReceived;      // 接收到数据报文明文事件
        public event EventHandler<TcpServerConnectedEventArgs> ServerConnected;                 // 与服务器的连接已建立事件
        public event EventHandler<TcpServerDisconnectedEventArgs> ServerDisconnected;           // 与服务器的连接已断开事件
        public event EventHandler<TcpServerExceptionOccurredEventArgs> ServerExceptionOccurred; // 与服务器的连接发生异常事件

        private TcpClient tcpClient;
        private bool disposed = false;
        private int retries = 0;        // 重连计数

        private readonly object _reconnectGate = new object();
        private bool _closing = false;
        private bool _isConnecting = false;
        private bool _isReconnecting = false;
        private Timer _reconnectTimer;

        public AsyncTcpClient(IPAddress remoteIPAddress, int remotePort)
        {
            this.Addresses = remoteIPAddress;
            this.Port = remotePort;
            this.Encoding = Encoding.Default;
            this.Retries = 3;
            this.RetryInterval = 5;
        }

        /// <summary>
        /// 是否已与服务器建立连接
        /// </summary>
        public bool Connected
        {
            get
            {
                try
                {
                    return this.tcpClient != null && this.tcpClient.Connected;
                }
                catch
                {
                    return false;
                }
            }
        }

        /// <summary>
        /// 远端服务器的IP地址列表
        /// </summary>
        public IPAddress Addresses { get; private set; }

        /// <summary>
        /// 远端服务器的端口
        /// </summary>
        public int Port { get; private set; }

        /// <summary>
        /// 连接重试次数
        /// </summary>
        public int Retries { get; set; }

        /// <summary>
        /// 连接重试间隔
        /// </summary>
        public int RetryInterval { get; set; }

        /// <summary>
        /// 远端服务器终结点
        /// </summary>
        public IPEndPoint RemoteIPEndPoint
        {
            get
            {
                return new IPEndPoint(this.Addresses, this.Port);
            }
        }

        /// <summary>
        /// 通信所使用的编码
        /// </summary>
        public Encoding Encoding { get; set; }
        uint on = 1;

        /// <summary>
        /// 连接到服务器
        /// </summary>
        /// <returns></returns>
        public AsyncTcpClient Connect()
        {
            lock (_reconnectGate)
            {
                _closing = false;
            }

            if (this.Connected)
            {
                return this;
            }

            this.ConnectInternal(resetRetries: true);
            return this;
        }

        private void ConnectInternal(bool resetRetries)
        {
            TcpClient client;

            lock (_reconnectGate)
            {
                if (_closing || disposed)
                {
                    return;
                }

                // Prevent multiple in-flight connection attempts.
                if (_isConnecting || _isReconnecting)
                {
                    return;
                }

                _isConnecting = true;
                if (resetRetries)
                {
                    retries = 0;
                }

                client = new TcpClient();
                this.tcpClient = client;
            }

            try
            {
                client.Client.IOControl(IOControlCode.KeepAliveValues, KeepAlive(1, 500, 500), null);
                client.BeginConnect(this.Addresses, this.Port, new AsyncCallback(this.HandleTcpServerConnected), client);
            }
            catch
            {
                try { client.Close(); } catch { }

                lock (_reconnectGate)
                {
                    _isConnecting = false;
                }

                if (!_closing && !disposed)
                {
                    ScheduleReconnect("connect begin failed");
                }
            }
        }

        private void HandleRemoteDisconnect(TcpClient client, string reason)
        {
            if (!ReferenceEquals(client, this.tcpClient))
            {
                return;
            }

            try { client.Close(); } catch { }

            this.RaiseServerDisconnected(this.Addresses, this.Port);

            if (!_closing && !disposed)
            {
                ScheduleReconnect(reason);
            }
        }

        private void ScheduleReconnect(string reason)
        {
            lock (_reconnectGate)
            {
                if (_closing || disposed)
                {
                    return;
                }

                if (this.Connected)
                {
                    return;
                }

                if (_isConnecting || _isReconnecting)
                {
                    return;
                }

                // Keep reconnecting until the client is closed/disposed.
                // Preserve the counter only for logging (avoid int overflow by wrapping).
                if (retries == int.MaxValue)
                {
                    retries = 0;
                }
                retries++;
                _isReconnecting = true;

                if (_reconnectTimer != null)
                {
                    try { _reconnectTimer.Dispose(); } catch { }
                    _reconnectTimer = null;
                }

                Diagnosis.Post($"[AsyncTcpClient] schedule reconnect attempt {retries}/{this.Retries} in {this.RetryInterval}s. Reason={reason}");

                Timer t = null;
                t = new Timer(_ =>
                {
                    try
                    {
                        lock (_reconnectGate)
                        {
                            _isReconnecting = false;
                            _reconnectTimer = null;
                        }

                        this.ConnectInternal(resetRetries: false);
                    }
                    catch { }
                    finally
                    {
                        try { t.Dispose(); } catch { }
                    }
                }, null, TimeSpan.FromSeconds((double)this.RetryInterval), Timeout.InfiniteTimeSpan);

                _reconnectTimer = t;
            }
        }

        private byte[] KeepAlive(int onOff, int keepAliveTime, int keepAliveInterval)
        {
            byte[] buffer = new byte[12];
            BitConverter.GetBytes(onOff).CopyTo(buffer, 0);
            BitConverter.GetBytes(keepAliveTime).CopyTo(buffer, 4);
            BitConverter.GetBytes(keepAliveInterval).CopyTo(buffer, 8);
            return buffer;
        }

        /// <summary>
        /// 关闭与服务器的连接
        /// </summary>
        /// <returns>异步TCP客户端</returns>
        public AsyncTcpClient Close()
        {
            TcpClient clientToClose = null;
            bool wasConnected = false;

            lock (_reconnectGate)
            {
                _closing = true;
                retries = 0;
                _isConnecting = false;
                _isReconnecting = false;

                if (_reconnectTimer != null)
                {
                    try { _reconnectTimer.Dispose(); } catch { }
                    _reconnectTimer = null;
                }

                clientToClose = this.tcpClient;
                wasConnected = clientToClose != null && clientToClose.Connected;
                this.tcpClient = null;
            }

            if (clientToClose != null)
            {
                try { clientToClose.Close(); } catch { }
            }

            if (wasConnected)
            {
                this.RaiseServerDisconnected(this.Addresses, this.Port);
            }
            return this;
        }

        private void HandleTcpServerConnected(IAsyncResult ar)
        {
            TcpClient client = (TcpClient)ar.AsyncState;
            try
            {
                if (!ReferenceEquals(client, this.tcpClient))
                {
                    // Stale connect callback for an old TcpClient instance.
                    try { client.Close(); } catch { }
                    return;
                }

                client.EndConnect(ar);
                this.RaiseServerConnected(this.Addresses, this.Port);

                lock (_reconnectGate)
                {
                    this.retries = 0;
                    _isConnecting = false;
                    _isReconnecting = false;

                    if (_reconnectTimer != null)
                    {
                        try { _reconnectTimer.Dispose(); } catch { }
                        _reconnectTimer = null;
                    }
                }

                byte[] buffer = new byte[client.ReceiveBufferSize];
                var state = new DatagramReadState { Client = client, Buffer = buffer };
                client.GetStream().BeginRead(buffer, 0, buffer.Length, new AsyncCallback(this.HandleDatagramReceived), state);
            }
            catch (Exception ex)
            {
                if (!ReferenceEquals(client, this.tcpClient))
                {
                    return;
                }

                lock (_reconnectGate)
                {
                    _isConnecting = false;
                }

                if (!_closing && !disposed)
                {
                    Diagnosis.Post($" HandleTcpServerConnected {this.Addresses} {this.Port} 连接断开,尝试重连...");
                    ScheduleReconnect("connect failed");
                }
            }
        }

        private void HandleDatagramReceived(IAsyncResult ar)
        {
            DatagramReadState state = null;
            TcpClient client = null;
            byte[] buffer = null;

            try
            {
                state = (DatagramReadState)ar.AsyncState;
                client = state.Client;
                buffer = state.Buffer;

                if (!ReferenceEquals(client, this.tcpClient))
                {
                    // Stale read callback for an old TcpClient instance.
                    return;
                }

                NetworkStream stream = client.GetStream();
                int numberOfReadBytes = 0;
                try
                {
                    numberOfReadBytes = stream.EndRead(ar);
                }
                catch
                {
                    numberOfReadBytes = 0;
                }

                if (numberOfReadBytes == 0)
                {
                    HandleRemoteDisconnect(client, "zero-byte read");
                    return;
                }

                byte[] receivedBytes = new byte[numberOfReadBytes];
                Buffer.BlockCopy(buffer, 0, receivedBytes, 0, numberOfReadBytes);
                this.RaiseDatagramReceived(client, receivedBytes);
                this.RaisePlaintextReceived(client, receivedBytes);

                // then start reading from the network again
                stream.BeginRead(buffer, 0, buffer.Length, new AsyncCallback(this.HandleDatagramReceived), state);
            }
            catch (Exception ex)
            {
                Trace.WriteLine(ex.Message);

                if (client != null && ReferenceEquals(client, this.tcpClient) && !_closing && !disposed)
                {
                    HandleRemoteDisconnect(client, "read failed");
                }
            }
        }

        private void RaiseDatagramReceived(TcpClient sender, byte[] datagram)
        {
            if (this.DatagramReceived != null)
            {
                this.DatagramReceived(this, new TcpDatagramReceivedEventArgs<byte[]>(sender, datagram));
            }
        }

        private void RaisePlaintextReceived(TcpClient sender, byte[] datagram)
        {
            if (this.PlaintextReceived != null)
            {
                //this.PlaintextReceived(this, new TcpDatagramReceivedEventArgs<string>(sender, this.Encoding.GetString(datagram, 0, datagram.Length)));
                this.PlaintextReceived(this, new TcpDatagramReceivedEventArgs<byte[]>(sender, datagram));
            }
        }

        private void RaiseServerConnected(IPAddress ipAddresses, int port)
        {
            if (this.ServerConnected != null)
            {
                this.ServerConnected(this, new TcpServerConnectedEventArgs(ipAddresses, port));
            }
        }

        private void RaiseServerDisconnected(IPAddress ipAddresses, int port)
        {
            if (this.ServerDisconnected != null)
            {
                this.ServerDisconnected(this, new TcpServerDisconnectedEventArgs(ipAddresses, port));
            }
        }

        private void RaiseServerExceptionOccurred(IPAddress ipAddresses, int port, Exception innerException)
        {
            if (this.ServerExceptionOccurred != null)
            {
                this.ServerExceptionOccurred(this, new TcpServerExceptionOccurredEventArgs(ipAddresses, port, innerException));
            }
        }

        /// <summary>
        /// 发送报文
        /// </summary>
        /// <param name="datagram"></param>
        public void Send(byte[] datagram)
        {
            if (datagram == null)
            {
                throw new ArgumentNullException("datagram");
            }
            if (!this.Connected)
            {
                this.RaiseServerDisconnected(this.Addresses, this.Port);
                throw new InvalidProgramException("This client has not connected to server.");
            }
            this.tcpClient.GetStream().BeginWrite(datagram, 0, datagram.Length, new AsyncCallback(this.HandleDatagramWritten), this.tcpClient);
        }

        private void HandleDatagramWritten(IAsyncResult ar)
        {
            ((TcpClient)ar.AsyncState).GetStream().EndWrite(ar);
        }

        public void Send(string datagram)
        {
            this.Send(this.Encoding.GetBytes(datagram));
        }

        /// <summary>
        /// 释放非托管资源
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!this.disposed)
            {
                this.disposed = true;

                if (disposing)
                {
                    try
                    {
                        this.Close();
                    }
                    catch// (SocketException ex)
                    {
                    }
                }
            }
        }

    }
}