bo/Workbench/QMonitor/qmonlib/QMonReceiver.cs

using System.Collections.Concurrent;
using System.IO.Abstractions;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.QMonitor.Interfaces;

namespace Quadarax.Foundation.Core.QMonitor
{
    public delegate void GeneralDataReceivedDelegate(MonGeneral general);
    public delegate void DataReceivedDelegate(MonData[] data);
    public class QMonReceiver : QMonHostBase<QMonReceiverConfiguration>
    {
        #region *** Properties ***
        protected override Uri UriGeneral => Configuration.SourceUriGeneral;
        protected override Uri UriData => Configuration.SourceUriData;
        protected override TimeSpan IntervalGeneral => Configuration.GeneralReceiverInterval;
        protected override TimeSpan? IntervalData => Configuration.DataCache.FlushCacheInterval;
        public MonGeneral[] AvailableGeneral => _generalCache.Values.Select(x => x.Data).ToArray();
        #endregion
        #region *** Fields ***
        private GeneralDataReceivedDelegate _generalReceivedCallback;
        private DataReceivedDelegate _dataReceivedCallback;
        private bool _isReceivingGeneral;
        private CancellationTokenSource _dataReceiverCancel;
        private IDictionary<string, MonReceiverGeneralCahedItem> _generalCache = new ConcurrentDictionary<string, MonReceiverGeneralCahedItem>();
        private IList<MonData> _dataCache = new List<MonData>();
        private JsonSerializerOptions _jsonSerializerOptions;
        private DateTime _lastSavedTimestamp;
        private IFileSystem _fileSystem;
        private bool _isDataClosing;
        #endregion
        #region *** Constructors ***
        public QMonReceiver(IFileSystem fileSystem, QMonReceiverConfiguration configuration, ILogger logger, GeneralDataReceivedDelegate generalDataReceivedCallback = null, DataReceivedDelegate liveDataReceivedCallback = null) : base(configuration, logger)
        {
            _fileSystem = fileSystem ?? throw new ArgumentNullException(nameof(fileSystem));
            _lastSavedTimestamp = DateTime.Now;
            _generalReceivedCallback = generalDataReceivedCallback;
            _dataReceivedCallback = liveDataReceivedCallback;
            _jsonSerializerOptions = new JsonSerializerOptions
            {
                IgnoreNullValues = true,
                WriteIndented = false,
                PropertyNameCaseInsensitive = true,
                PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
                MaxDepth = 10
            };
            _jsonSerializerOptions.Converters.Add(new JsonStringEnumConverter());
            Configuration.DataCache.Path = Configuration.DataCache.Path.Replace("%AppData%",
                Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData));
            if (!_fileSystem.Directory.Exists(Configuration.DataCache.Path))
            {
                _fileSystem.Directory.CreateDirectory(Configuration.DataCache.Path);
                Log(LogSeverityEnum.Debug, $"Data receiver created data cache directory '{Configuration.DataCache.Path}'");

            }
        }
        #endregion

        #region *** Public Operations ***
        public IReceiverHandler CreateHandler(string instanceIdentifier, string name)
        {
            return null;
        }
        #endregion

        #region *** Private overrides ***

        protected override void OnOpen()
        {
            _isDataClosing = false;
            var _dataReceiverCancel = new CancellationTokenSource();
            new System.Threading.Thread(() =>
            {
                try
                {
                    Log(LogSeverityEnum.Debug, "Data receiver thread started.");
                    while (!_dataReceiverCancel.Token.IsCancellationRequested)
                        OnThreadData(_dataReceiverCancel.Token);
                }
                catch (OperationCanceledException) 
                {
                    Log(LogSeverityEnum.Info, $"Data receiver thread on {UriData} unexpected stopped");
                }
                Log(LogSeverityEnum.Debug, "Data receiver thread stoped.");
            }).Start();
            Log(LogSeverityEnum.Info, $"Data receiver thread on {UriData} started");
        }

        protected override void OnClose()
        {
            _isDataClosing = true;
            _dataReceiverCancel?.Cancel();
            Log(LogSeverityEnum.Info, $"Data receiver thread on {UriData} stopped");
        }

        protected override void OnTimerData(object? state)
        {
            _lastSavedTimestamp = DateTime.Now;
            var itemsToSave = _dataCache.Where(x => x.Timestamp < _lastSavedTimestamp).ToArray();
            var instances = itemsToSave.Select(x => x.InstanceIdentifier).Distinct();

            foreach (var instance in instances)
            {
                var dirName = _fileSystem.Path.Combine(Configuration.DataCache.Path, instance);
                if (!_fileSystem.Directory.Exists(dirName))
                {
                    _fileSystem.Directory.CreateDirectory(dirName);
                    Log(LogSeverityEnum.Debug, $"Data receiver created data cache directory '{dirName}'");
                }

                var batchData = itemsToSave.Where(x => x.InstanceIdentifier == instance).OrderBy(x => x.Timestamp).ToArray();
                _dataReceivedCallback?.Invoke(batchData);

                var dfrom = batchData.Min(x => x.Timestamp).ToFileTimeUtc();
                var dto = batchData.Max(x => x.Timestamp).ToFileTimeUtc();
                var sbFileName = new StringBuilder();
                sbFileName.Append("mondata_").Append(dfrom).Append("-").Append(dto).Append(".json");
                var fileName = _fileSystem.Path.Combine(dirName, sbFileName.ToString());
                using (var streamWriter = File.CreateText(fileName))
                {
                    JsonSerializer.Serialize(streamWriter.BaseStream, batchData, _jsonSerializerOptions);
                    streamWriter.Flush();
                }
                Log(LogSeverityEnum.Trace, $"Data receiver created dump data file '{fileName}'");
            }

            foreach (var item in itemsToSave)
                _dataCache.Remove(item);
        }

        protected override void OnTimerGeneral(object? state)
        {
            if (state == null) throw new ArgumentNullException(nameof(state));
            var owner = (QMonReceiver)state;
            if (owner._isReceivingGeneral) return;
            owner._isReceivingGeneral = true;
            try
            {
                var remoteEndPoint =
                    new IPEndPoint(IPAddress.Parse(owner.UriGeneral.DnsSafeHost), owner.UriGeneral.Port);
                var data = UdpClientGeneral.Receive(ref remoteEndPoint);
                var general = (MonGeneral)Binder.LoadFromString(Encoding.UTF8.GetString(data), typeof(MonGeneral));
                Log(LogSeverityEnum.Debug,
                    $"[ReceiverGeneral] receive general data from '{remoteEndPoint}' [{general.InstanceIdentifier}] size={data.Length} bytes.");

                _generalCache.TryAdd(general.InstanceIdentifier,
                    new MonReceiverGeneralCahedItem(DateTime.Now, remoteEndPoint, general));
                _generalReceivedCallback?.Invoke(general);
            }
            catch (SocketException e)
            {
                if (e.ErrorCode!= 10004) Log(LogSeverityEnum.Error, $"[ReceiverGeneral] Error receiving general data", e);
            }
            catch (Exception e)
            {
                Log(LogSeverityEnum.Error, $"[ReceiverGeneral] Error receiving general data", e);
            }
            finally
            {
                owner._isReceivingGeneral = false;
            }
        }

        private void OnThreadData(CancellationToken cancellation)
        {
            if (cancellation == null) throw new ArgumentNullException(nameof(cancellation));

            if (cancellation.IsCancellationRequested)
            {
                Log(LogSeverityEnum.Debug, $"[ReceiverData] Cancellation token detected on listener '{UriData}'.");
                return;
            }
            try
            {
                System.Threading.Thread.CurrentThread.Join(100);
                if (_isDataClosing) return;
                if (UdpClientData.Available == 0) return;
                var remoteEndPoint = new IPEndPoint(IPAddress.Parse(UriData.DnsSafeHost), UriData.Port);
                Log(LogSeverityEnum.Debug, $"[ReceiverData] Waiting for data on '{UriData}'...");
                var data = UdpClientData.Receive(ref remoteEndPoint);
                var monData = JsonSerializer.Deserialize<MonData>(data, _jsonSerializerOptions);
                if (monData == null)
                {
                    Log(LogSeverityEnum.Warn, $"[ReceiverData] cannot deserialize monitor data from '{remoteEndPoint}' size={data.Length} bytes. Skipped.");
                    return;
                }
                _dataCache.Add(monData);
                Log(LogSeverityEnum.Debug, $"[ReceiverData] receive monitor data from '{remoteEndPoint}' size={data.Length} bytes.");
            }
            catch (Exception e)
            {
                Log(LogSeverityEnum.Error, $"[ReceiverData] Error receiving monitor data", e);
            }
        }

        protected override UdpClient CreateUdpClient(Uri targetUri)
        {
            var udpClient = new UdpClient(targetUri.Port);
            return udpClient;
        }

        protected override void OnDisposing()
        {
            _dataReceiverCancel?.Cancel();
            _dataReceiverCancel?.Dispose();
            _dataReceiverCancel = null;
            base.OnDisposing();
        }


        #endregion


    }
}