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

using Edge.Core.Parser;
using Edge.Core.Parser.BinaryParser.MessageEntity;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Timer = System.Threading.Timer;
using System.Collections.Concurrent;
namespace Edge.Core.Processor
{
    /// <summary>
    /// With a build in timer to timely send msg from queue, 
    /// so every messages written will be firstly queued here and then delayed to send.
    /// NOTE, 
    /// </summary>
    /// <typeparam name="TRaw"></typeparam>
    /// <typeparam name="TMessage"></typeparam>
    public class TimeWindowWithActivePollingOutgoing<TRaw, TMessage> : Outgoing<TRaw, TMessage>, IDisposable where TMessage : MessageBase
    {
        static NLog.Logger logger = NLog.LogManager.LoadConfiguration("nlog.config").GetLogger("Communicator");
        //use dedicated thread for make sure the polling interval accurate
        private Thread pollingThread;
        private bool disposed = false;
        /// <summary>
        /// I've see a case that underlying communicator suddenly muted for seconds(maybe remote device got stuck inside??), 
        /// and later, bunch of bytes flushed in and got deserialzed into Messages and all were piped into Handlers, 
        /// which surely will trigger the handler frequently call the Write(...) for a bunch of responses, thus full fill this queue very quickly.
        /// </summary>
        private const int maxQueueLength = 200;
        //private int activelyPollingInterval = 600;

        private ConcurrentQueue<object> outgoingMessageQueue = new ConcurrentQueue<object>();

        public Func<TMessage> PollingMsgProducer { get; set; }

        /// <summary>
        /// by ms, Gets the polling interval for internal polling timer.
        /// </summary>
        public int PollingInterval { get; }
        //public TimeWindowWithActivePollingOutgoing(IIncoming<TMessage> incoming, int activelyPollingInterval) : base(incoming)
        //{
        //    this.PollingInterval = activelyPollingInterval;
        //    this.polling = new Timer((__) =>
        //    {
        //        var result = this.polling.Change(Timeout.Infinite, Timeout.Infinite);
        //        if (!result)
        //            logger.Warn("Polling timer failed to Change to Infinite!!!");
        //        if (this.outgoingMessageQueue.TryDequeue(out object pendingForWrite))
        //        {
        //            if (pendingForWrite is TMessage)
        //            {
        //                base.Write(pendingForWrite as TMessage);
        //            }
        //            else if (pendingForWrite is Tuple<TMessage, Func<TMessage, TMessage, bool>, Action<TMessage, TMessage>, int>)
        //            {
        //                var ___ = pendingForWrite as Tuple<TMessage, Func<TMessage, TMessage, bool>, Action<TMessage, TMessage>, int>;
        //                base.WriteAsync(___.Item1, ___.Item2, ___.Item3, ___.Item4);
        //            }
        //            else if (pendingForWrite is Tuple<TMessage, object>)
        //            {
        //                var ___ = pendingForWrite as Tuple<TMessage, object>;
        //                base.Write(___.Item1, ___.Item2);
        //            }
        //        }
        //        else
        //            if (this.PollingMsgProducer != null) { var m = this.PollingMsgProducer(); if (m != null) base.Write(m); }
        //        result = this.polling.Change(activelyPollingInterval, activelyPollingInterval);
        //        if (!result)
        //            logger.Warn("Polling timer failed to Change to activelyPollingInterval!!!");
        //    }, null, 2000, activelyPollingInterval);
        //}

        public TimeWindowWithActivePollingOutgoing(IIncoming<TMessage> incoming, int activelyPollingInterval, IServiceProvider services) : base(incoming, services)
        {
            this.PollingInterval = activelyPollingInterval;
            this.pollingThread = new Thread(async () =>
            {
                while (!this.disposed)
                {
                    await Task.Delay(this.PollingInterval);
                    if (this.outgoingMessageQueue.TryDequeue(out object pendingForWrite))
                    {
                        if (pendingForWrite is TMessage)
                        {
                            base.Write(pendingForWrite as TMessage);
                        }
                        else if (pendingForWrite is Tuple<TMessage, Func<TMessage, TMessage, bool>, Action<TMessage, TMessage>, int>)
                        {
                            var ___ = pendingForWrite as Tuple<TMessage, Func<TMessage, TMessage, bool>, Action<TMessage, TMessage>, int>;
                            base.WriteAsync(___.Item1, ___.Item2, ___.Item3, ___.Item4);
                        }
                        else if (pendingForWrite is Tuple<TMessage, object>)
                        {
                            var ___ = pendingForWrite as Tuple<TMessage, object>;
                            base.Write(___.Item1, ___.Item2);
                        }
                    }
                    else
                        if (this.PollingMsgProducer != null) { var m = this.PollingMsgProducer(); if (m != null) base.Write(m); }
                }
            });
            this.pollingThread.Start();
            //result = this.polling.Change(activelyPollingInterval, activelyPollingInterval);
            //if (!result)
            //    logger.Warn("Polling timer failed to Change to activelyPollingInterval!!!");
            //}, null, 2000, activelyPollingInterval);
        }

        private void HandleQueueMaxLengthReachedError()
        {
            if (this.outgoingMessageQueue.Count > maxQueueLength / 3)
            {
                logger.Info("TimeWindowWithActivePollingOutgoing, queue reached its 1/3 capacity, just inform and will do nothing.");
            }
            else if (this.outgoingMessageQueue.Count > maxQueueLength)
            {
                /* let's dump inside elements to see what happened.*/
                string logStr = "";

                var msgLogStrings = this.outgoingMessageQueue.OfType<TMessage>().Select(s => s.ToLogString());
                if (msgLogStrings.Any()) logStr += msgLogStrings.Aggregate((acc, n) => acc + "\r\n====next=======\r\n" + n);

                var tuple4Strings = this.outgoingMessageQueue.OfType<Tuple<TMessage, Func<TMessage, TMessage, bool>, Action<TMessage, TMessage>, int>>().Select(s => s.Item1.ToLogString());
                if (tuple4Strings.Any()) logStr += tuple4Strings.Aggregate((acc, n) => acc + "\r\n" + n);

                var tuple2Strings = this.outgoingMessageQueue.OfType<Tuple<TMessage, object>>().Select(s => s.Item1.ToLogString());
                if (tuple2Strings.Any()) logStr += tuple2Strings.Aggregate((acc, n) => acc + "\r\n" + n);

                /*I have no better way to handle this case but clear 1/3 of(oldest items) the queue*/
                for (int i = 0; i < maxQueueLength / 3; i++)
                {
                    object _;
                    this.outgoingMessageQueue.TryDequeue(out _);
                }
                //this.outgoingMessageQueue.Clear();
                logger.Error("TimeWindowWithActivePollingOutgoing, Will clear 1/3 of the queue since Max message queue length: " + maxQueueLength
                    + " reached, elements are: " + logStr);
            }
        }

        /// <summary>
        /// Write a msg to queue, and actual write action will be performed on another thread.
        /// </summary>
        /// <param name="message"></param>
        public override void Write(TMessage message)
        {
            this.outgoingMessageQueue.Enqueue(message);
            this.HandleQueueMaxLengthReachedError();
        }

        /// <summary>
        /// actual write action will be performed on another thread.
        /// </summary>
        /// <param name="request"></param>
        /// <param name="responseCapture"></param>
        /// <param name="callback"></param>
        /// <param name="timeout"></param>
        public override void WriteAsync(TMessage request, Func<TMessage, TMessage, bool> responseCapture, Action<TMessage, TMessage> callback, int timeout)
        {
            this.outgoingMessageQueue.Enqueue(new Tuple<TMessage, Func<TMessage, TMessage, bool>, Action<TMessage, TMessage>, int>(request, responseCapture, callback, timeout));
            this.HandleQueueMaxLengthReachedError();
        }

        /// <summary>
        /// actual write action will be performed on another thread.
        /// </summary>
        /// <param name="message"></param>
        /// <param name="extraControlParameter"></param>
        public override void Write(TMessage message, object extraControlParameter)
        {
            this.outgoingMessageQueue.Enqueue(new Tuple<TMessage, object>(message, extraControlParameter));
            this.HandleQueueMaxLengthReachedError();
        }

        public void Dispose()
        {
            this.disposed = true;
        }
    }
}