FCCLite/Edge.Core/Processor/Outgoing/Outgoing.cs

using Edge.Core.Parser;
using Edge.Core.Parser.BinaryParser.MessageEntity;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Edge.Core.Processor
{
    public class Outgoing<TRaw, TMessage> : IOutgoing<TRaw, TMessage> where TMessage : MessageBase
    {
        //static NLog.Logger logger = NLog.LogManager.LoadConfiguration("nlog.config").GetLogger("Communicator");
        static ILogger logger = NullLogger.Instance;
        protected IIncoming<TMessage> incoming;
        public event EventHandler<OutgoingEventArg<TMessage>> OnWriting;

        //接收当前连接是否正常
        private TaskCompletionSource<bool> isConnectTask;

        public Outgoing(IIncoming<TMessage> incoming)
        {
            this.incoming = incoming;
        }

        public Outgoing(IIncoming<TMessage> incoming, IServiceProvider services)
        {
            this.incoming = incoming;
            if (services != null)
            {
                var loggerFactory = services.GetRequiredService<ILoggerFactory>();
                logger = loggerFactory.CreateLogger("Communicator");
            }
        }

        /// <summary>
        /// 通知当前链接已连接
        /// </summary>
        public void OnConnect()
        {
            isConnectTask = new TaskCompletionSource<bool>();
        }

        /// <summary>
        /// 通知当前链接已断开
        /// </summary>
        public void OnDisconnect()
        {
            isConnectTask.SetResult(false);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="request">request to remote peer, which should trigger a response from remote peer.</param>
        /// <param name="responseCapture">predict of response correlated to request, first is the request, 2nd is the pending for capture response</param>
        /// <param name="callback">when response catpured, or timed, this callback will be called</param>
        /// <param name="timeout">time out, by ms</param>
        public virtual void WriteAsync(TMessage request, Func<TMessage, TMessage, bool> responseCapture, Action<TMessage, TMessage> callback, int timeout)
        {
            int isCallbackCalled = 0;
            EventHandler eventHandler = null;
            eventHandler = (s, a) =>
            {
                var incoming = s as Incoming<TMessage>;
                if (responseCapture(request, incoming.Message))
                {
                    this.incoming.OnMessageIncoming -= eventHandler;
                    if (Interlocked.CompareExchange(ref isCallbackCalled, 1, 0) == 0)
                    {
                        this.incoming.DisablePropagate = true;
                        try
                        {
                            callback(request, incoming.Message);
                        }
                        catch (Exception exx) { logger.LogError("Outgoing.WriteAsync(...,...,...,...) exceptioned in callback, detail: " + exx); }
                    }
                }
            };
            this.incoming.OnMessageIncoming += eventHandler;

            var _ = new System.Timers.Timer(timeout);
            _.Elapsed += (__, ___) =>
            {
                _.Stop();
                this.incoming.OnMessageIncoming -= eventHandler;
                if (Interlocked.CompareExchange(ref isCallbackCalled, 1, 0) == 0)
                {
                    callback(request, null);
                }
            };
            _.Start();
            var safe = this.OnWriting;
            safe?.Invoke(this, new OutgoingEventArg<TMessage>() { Message = request });
        }

        public virtual Task<TMessage> WriteAsync(TMessage request, Func<TMessage, TMessage, bool> responseCapture, int timeout)
        {
            var tcs = new TaskCompletionSource<TMessage>();
            this.WriteAsync(request, responseCapture, (oriRequest, response) =>
            {
                if (response != null)
                    tcs.SetResult(response);
                else
                    tcs.SetResult(null);
            }, timeout);
            return tcs.Task;
        }

        public virtual async Task<WriteRepsonse> WriteAsyncAndCheckIsConnect(TMessage request, Func<TMessage, TMessage, bool> responseCapture, int timeout)
        {
            if(isConnectTask == null) return new WriteRepsonse()
            {
                ResponseType = WriteResponseType.DISCONNECT,
                Data = false,
            };
            Task<TMessage> sendTask = WriteAsync(request, responseCapture, timeout);
            Task responseTask = await Task.WhenAny(sendTask, isConnectTask.Task);
            var type = WriteResponseType.TIME_OUT;
            object? result = null;
            if (responseTask == sendTask)
            {
                result = await sendTask;
                if (result != null) type = WriteResponseType.OK;
            }
            if(responseTask == isConnectTask.Task)
            {
                result = await isConnectTask.Task;
                type = WriteResponseType.DISCONNECT;
            }

            return new WriteRepsonse()
            {
                ResponseType = type,
                Data = result,
            };
        }

        public virtual void Write(TMessage message)
        {
            var safe = this.OnWriting;
            safe?.Invoke(this, new OutgoingEventArg<TMessage>() { Message = message });
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <param name="redirectParameter"></param>
        public virtual void Write(TMessage message, object extraControlParameter)
        {
            var safe = this.OnWriting;
            safe?.Invoke(this, new OutgoingEventArg<TMessage>() { Message = message, ExtraControlParameter = extraControlParameter });
        }
    }

    public class WriteRepsonse
    {
        public WriteResponseType ResponseType { get; set; }

        public object? Data { get; set; }
    }

    public enum WriteResponseType
    {
        TIME_OUT,
        DISCONNECT,
        OK
    }
}