FCCLite/Edge.Core/Processor/Communicator/TcpServerCommunicator.cs

using Edge.Core.Parser.BinaryParser.MessageEntity;
using Edge.Core.Parser.BinaryParser.Util;
using Edge.Core.Processor.Dispatcher.Attributes;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Edge.Core.Processor.Communicator
{
    [MetaPartsDescriptor(
        "lang-zh-cn:Tcp(本程序为服务器端)通讯器lang-en-us:Tcp(as server) communicator",
        "lang-zh-cn:基于TCP/IP技术的通讯器, FC作为服务器端等待客户端连接lang-en-us:TCP/IP based communicator, FC as the server and wait for connections")]
    public class TcpServerCommunicator<T> : ICommunicator<byte[], T> where T : MessageTemplateBase
    {
        private CancellationTokenSource readAsyncCancellationTokenSource;
        private DateTime? lastReceiveMsgDataFromTcpClientDateTime;
        private System.Timers.Timer clientSideActiveDetectionByCheckingLongTimeNoSeeDataIncomingWatchTimer;
        //protected static NLog.Logger logger = NLog.LogManager.LoadConfiguration("nlog.config").GetLogger("Communicator");
        protected ILogger logger = NullLogger.Instance;
        public static int tcpReceiveBufferSize = 1500;

        private Edge.Core.Parser.BinaryParser.ParserBase parser;

        private int localTcpServerListeningPort;
        protected TcpListener tcpListener;
        private TcpClient exclusiveTcpClient;
        private string exclusiveTcpClient_ClientRemoteEndPoint_Str = "?:?";
        //private bool isTcpConnBrokenDuringWrite = false;
        private object syncObject = new object();
        private object syncObject_Feed = new object();
        /// <summary>
        /// 0 for not started, 1 for started.
        /// </summary>
        private int isStarted = 0;
        protected bool disposed = false;
        private IMessageCutter<byte[]> messageCutter;

        //private List<MessageTemplateBase> continueParseStack = new List<MessageTemplateBase>();

        public event EventHandler OnConnected;
        public event EventHandler OnDisconnected;

        public event EventHandler<CommunicatorErrorMessageReadEventArg> OnErrorMessageRead;
        public event EventHandler<CommunicatorEventArg<byte[], T>> OnRawDataWriting;
        public event EventHandler<CommunicatorEventArg<byte[], T>> OnDataReceived;

        /// <summary>
        /// fired once raw data read from underlying tcp connection, without message cutting yet.
        /// </summary>
        protected Action<byte[]> OnRawDataReceived;

        public string Identity { get; set; }

        [ParamsJsonSchemas("TcpServerCommunicatorCtorSchema")]
        public TcpServerCommunicator(IMessageCutter<byte[]> binaryMsgCutter,
            Edge.Core.Parser.BinaryParser.ParserBase parser,
            int localTcpServerTcpListeningPortNumber,
            int enableClientSideActiveDetection,
            string enableClientSideIdentityRegistration,
            string enableClientSideConnLevelHeartbeat,
            string customLoggerFileName,
            IServiceProvider services)
        {
            this.Identity = "*:" + localTcpServerTcpListeningPortNumber;
            if (services != null)
            {
                var loggerFactory = services.GetRequiredService<ILoggerFactory>();
                this.logger = loggerFactory.CreateLogger("Communicator");
                if (!string.IsNullOrEmpty(customLoggerFileName))
                    if (customLoggerFileName == "*")
                        this.logger = loggerFactory.CreateLogger("DynamicPrivate_Comm_" + localTcpServerTcpListeningPortNumber);
                    else
                        this.logger = loggerFactory.CreateLogger("DynamicPrivate_Comm_" + customLoggerFileName);
            }

            this.messageCutter = binaryMsgCutter;

            this.parser = parser;
            //this.remoteTcpServerListeningIpAddress = localTcpServerListeningIpAddress;
            this.localTcpServerListeningPort = localTcpServerTcpListeningPortNumber;
            if (this.messageCutter != null)
            {
                this.messageCutter.OnInvalidMessageRead += (____, ______) =>
                {
                    var loggingStr = $"Bytes msg from tcp client: {exclusiveTcpClient_ClientRemoteEndPoint_Str} Read Invalid data, detail: {(______?.Message ?? "")}";
                    this.logger.LogInformation(loggingStr);
                    this.OnErrorMessageRead?.Invoke(this, new CommunicatorErrorMessageReadEventArg(null, loggingStr));
                };
                this.messageCutter.OnMessageCut += (s, _) =>
                {
                    this.lastReceiveMsgDataFromTcpClientDateTime = DateTime.Now;
                    var eventArg = new CommunicatorEventArg<byte[], T>();
                    try
                    {
                        eventArg.Data = this.messageCutter.Message;
                        eventArg.Message = this.parser.Deserialize(this.messageCutter.Message.ToArray()) as T;
                        if (logger.IsEnabled(LogLevel.Debug))
                            this.logger.LogDebug("      Parsed: " + eventArg.Message.ToLogString());
                    }
                    catch (Exception ex)
                    {
                        var loggingStr = "Message from " + exclusiveTcpClient_ClientRemoteEndPoint_Str
                            + " exceptioned in deserilaizing bytes:\r\n 0x" + this.messageCutter.Message.ToHexLogString() + "\r\n exception detail:\r\n" + ex;
                        this.logger.LogError(loggingStr);
                        this.OnErrorMessageRead?.Invoke(this, new CommunicatorErrorMessageReadEventArg(this.messageCutter.Message, loggingStr));
                        return;
                    }

                    try
                    {
                        this.OnDataReceived?.Invoke(this, eventArg);
                    }
                    catch (Exception ex)
                    {
                        this.logger.LogError("Message from " + exclusiveTcpClient_ClientRemoteEndPoint_Str
                            + " exceptioned in handle message:\r\n" + eventArg.Message.ToLogString() + "\r\n exceptioned detail: \r\n" + ex);
                    }
                };
            }

            if (enableClientSideActiveDetection > 0)
            {
                this.clientSideActiveDetectionByCheckingLongTimeNoSeeDataIncomingWatchTimer = new System.Timers.Timer();
                this.clientSideActiveDetectionByCheckingLongTimeNoSeeDataIncomingWatchTimer.Interval = 2000;
                this.clientSideActiveDetectionByCheckingLongTimeNoSeeDataIncomingWatchTimer.Elapsed += (s, a) =>
                {
                    //null indicates this communicator just started
                    if (this.lastReceiveMsgDataFromTcpClientDateTime == null) return;
                    if (DateTime.Now.Subtract(this.lastReceiveMsgDataFromTcpClientDateTime ?? DateTime.MinValue).TotalSeconds >= enableClientSideActiveDetection
                        && this.exclusiveTcpClient != null)
                    {
                        this.logger.LogInformation($"Long time no see data from tcp client with: { this.exclusiveTcpClient_ClientRemoteEndPoint_Str }, will actively disconnect it...");
                        this.readAsyncCancellationTokenSource?.Cancel();
                    }
                };
                this.clientSideActiveDetectionByCheckingLongTimeNoSeeDataIncomingWatchTimer.Start();
            }
        }

        public void Dispose()
        {
            try
            {
                this.readAsyncCancellationTokenSource?.Cancel();
            }
            catch { }
            this.isStarted = 0;
            this.tcpListener?.Stop();
            this.clientSideActiveDetectionByCheckingLongTimeNoSeeDataIncomingWatchTimer?.Stop();
            this.disposed = true;
        }

        public virtual async Task<bool> Stop()
        {
            this.Dispose();
            return true;
        }

        public virtual Task<bool> Start()
        {
            if (0 == Interlocked.CompareExchange(ref this.isStarted, 1, 0))
            {
                try
                {
                    this.tcpListener = new TcpListener(IPAddress.Any, this.localTcpServerListeningPort);
                    this.tcpListener.Start();
                    this.logger.LogInformation($"TcpListener listened on localPort: {this.localTcpServerListeningPort}");
                    var _ = Task.Run(async () =>
                    {
                        while (this.isStarted == 1)
                        {
                            logger.LogInformation($"Waitting for connection on localPort: {this.localTcpServerListeningPort}");
                            var newTcpClient = await this.tcpListener.AcceptTcpClientAsync();
                            this.logger.LogInformation($"   A tcp client with remote ip/port: {newTcpClient.Client.RemoteEndPoint} has connected in");

                            if (this.exclusiveTcpClient != null)
                            {
                                logger.LogInformation($"    There's already a previous TcpClient established as exclusive, so close this new one with remote ip/port: {newTcpClient.Client.RemoteEndPoint}");
                                try
                                {
                                    newTcpClient.Close();
                                }
                                catch { }
                                continue;
                            }

                            this.exclusiveTcpClient = newTcpClient;
                            this.exclusiveTcpClient_ClientRemoteEndPoint_Str = this.exclusiveTcpClient.Client.RemoteEndPoint.ToString();
                            this.logger.LogInformation($"   Tcp client with remote ip/port: {this.exclusiveTcpClient_ClientRemoteEndPoint_Str} has been chosen as exclusive");
                            var ___ = Task.Run(async () =>
                            {

                                #region try set the tcp client with TCP keepalive feature, but seems does not work.

                                if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
                                {
                                }
                                else
                                {
                                    //logger.LogInformation($"Enabled tcp keep alive for tcpClient");
                                    //overall switch to enable the feature.
                                    this.exclusiveTcpClient.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
                                    //when entering tcp idle, how much time wait before send a KeepAlive package.
                                    this.exclusiveTcpClient.Client.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, 20);
                                    //when in tcp idle, and failed to received previous KeepAlive response, sequence of KeepAlive packages will send by this interval, until response received, or TcpKeepAliveRetryCount reached.
                                    this.exclusiveTcpClient.Client.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, 8);
                                    //when in tcp idle, how many KeepAlive package response have not received will treat as tcp broken, and trigger tcp disconnected exception.
                                    this.exclusiveTcpClient.Client.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, 3);
                                }

                                #endregion

                                this.lastReceiveMsgDataFromTcpClientDateTime = DateTime.Now;
                                this.OnConnected?.Invoke(this, null);
                                while (this.isStarted == 1)
                                {
                                    this.readAsyncCancellationTokenSource = new CancellationTokenSource();
                                    byte[] buffer = new byte[tcpReceiveBufferSize];
                                    int bytesReadCount;
                                    try
                                    {
                                        bytesReadCount = await this.exclusiveTcpClient.GetStream().ReadAsync(buffer, this.readAsyncCancellationTokenSource.Token);
                                        if (bytesReadCount == 0)
                                            throw new Exception("tcp server received 0 count data indicates the connection is broken");
                                        if (logger.IsEnabled(LogLevel.Debug))
                                            this.logger.LogDebug($"TCP from { this.exclusiveTcpClient_ClientRemoteEndPoint_Str } <---Incoming: 0x" + buffer.Take(bytesReadCount).ToHexLogString());
                                    }
                                    catch (Exception eeee)
                                    {
                                        logger.LogError($"tcp client with remote ip/port: {this.exclusiveTcpClient_ClientRemoteEndPoint_Str} exceptioned in GetStream().ReadAsync(), treat as tcp disconnection, detail: {eeee}");
                                        try
                                        {
                                            this.exclusiveTcpClient.Close();
                                        }
                                        finally { this.exclusiveTcpClient = null; }

                                        try
                                        {
                                            this.OnDisconnected?.Invoke(this, null);
                                        }
                                        catch { }
                                        break;
                                    }

                                    var data = buffer.Take(bytesReadCount).ToArray();
                                    try
                                    {
                                        this.OnRawDataReceived?.Invoke(data);
                                    }
                                    catch { }
                                    try
                                    {
                                        lock (this.syncObject_Feed)
                                        {
                                            this.messageCutter.Feed(data,  this.localTcpServerListeningPort);
                                        }
                                    }
                                    catch (Exception ex)
                                    {
                                        this.logger.LogError($"Exception in Parsing msg bytes: 0x{ buffer.Take(bytesReadCount).ToHexLogString()}, detail: {Environment.NewLine}{ ex.ToString()}");
                                    }
                                }
                            });
                        }
                    });

                    return Task.FromResult(true);
                }
                catch (Exception exxx)
                {
                    logger.LogError($"Start tcp listener on port: {this.localTcpServerListeningPort} exceptioned: {exxx}");
                    return Task.FromResult(false);
                }
            }

            return Task.FromResult(false);
        }

        public bool Write(T message)
        {
            if (this.exclusiveTcpClient == null)
            {
                //if (this.logger.IsEnabled(LogLevel.Trace))
                //    this.logger.LogTrace($"Write failed as no tcp client connected in yet");
                return false;
            }

            if (message == null) return false;
            byte[] rawData;
            try
            {
                rawData = this.parser.Serialize(message);
                var arg = new CommunicatorEventArg<byte[], T>() { Data = rawData, Message = message, Continue = true };
                this.OnRawDataWriting?.Invoke(this, arg);
                if (this.exclusiveTcpClient == null || !arg.Continue) { this.logger.LogError("Write failed, this.tcpClient is null: " + (this.exclusiveTcpClient is null)); return false; }
            }
            catch (Exception exx)
            {
                var msgLogStr = "";
                try
                {
                    msgLogStr = message.ToLogString();
                }
                catch
                {
                    msgLogStr = "exceptioned for get ToLogString()";
                }

                this.logger.LogError("Tcp Write failed in serialize or event raise for msg: " + message.GetType() + " -> " + msgLogStr + "\r\n detail: " + exx);
                return false;
            }

            lock (this.syncObject)
            {
                try
                {
                    if (logger.IsEnabled(LogLevel.Debug))
                        this.logger.LogDebug("TCP to " + (this.exclusiveTcpClient_ClientRemoteEndPoint_Str) + " Outgoing--->: " + message.ToLogString() + "\r\n     0x" + rawData.ToHexLogString());
                    var sendCount = this.exclusiveTcpClient.Client.Send(rawData);
                    if (sendCount == 0)
                        throw new InvalidOperationException("the send count in this.exclusiveTcpClient.Client.Send is 0");
                }
                catch (Exception exx)
                {
                    this.logger.LogError("Send tcp msg to "
                        + (this.exclusiveTcpClient_ClientRemoteEndPoint_Str) + " Write(...) exceptioned, treat as a broken tcp connection, will cancel the data read as well, detail: " + exx);
                    try
                    {
                        this.readAsyncCancellationTokenSource.Cancel();
                    }
                    catch { }
                    this.OnDisconnected?.Invoke(this, null);
                }
            }

            return true;
        }

        public bool Write(T message, object extraControlParameter)
        {
            throw new NotImplementedException();
        }
    }
}