bo/Workbench/QMonitor/qmonlib/QMonReceiver.cs

using System.Collections.Concurrent;
using System.Collections.Immutable;
using System.Diagnostics;
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.Collections;
using Quadarax.Foundation.Core.Json;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.QMonitor.Interfaces;
using Quadarax.Foundation.Core.Thread;

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 readonly GeneralDataReceivedDelegate? _generalReceivedCallback;
        private readonly DataReceivedDelegate? _dataReceivedCallback;
        private readonly IDictionary<string, MonReceiverGeneralCahedItem> _generalCache = new ConcurrentDictionary<string, MonReceiverGeneralCahedItem>();
        private readonly IList<MonData> _dataCache = new LockedList<MonData>();
        private readonly JsonSerializerOptions _jsonSerializerOptions;
        private long _lastSavedTimestamp;
        private readonly IFileSystem _fileSystem;
        private ContiguousWorker? _dataReceiverWorker;
        private ILogger? _logger;
        #endregion
        #region *** Constructors ***
        public QMonReceiver(IFileSystem fileSystem, QMonReceiverConfiguration configuration,bool purgeOnInit, ILogger logger, GeneralDataReceivedDelegate? generalDataReceivedCallback = null, DataReceivedDelegate? liveDataReceivedCallback = null) : base(configuration, logger)
        {
            _logger = logger;
            _fileSystem = fileSystem ?? throw new ArgumentNullException(nameof(fileSystem));
            _lastSavedTimestamp = DateTime.Now.ToFileTimeUtc();
            _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}'");

            }

            if (purgeOnInit)
            {
                var files = _fileSystem.Directory.GetFiles(Configuration.DataCache.Path, "*.json", SearchOption.AllDirectories);
                var deletedCnt = 0;
                foreach (var file in files)
                {
                    _fileSystem.File.Delete(file);
                    deletedCnt++;
                }
                Log(LogSeverityEnum.Info, $"Data receiver purge data cache on init, removes {deletedCnt} files");
            }
        }
        #endregion

        #region *** Public Operations ***
        public IReceiverHandler CreateHandler(string instanceIdentifier, string name,TimeSpan refreshInterval,string? alternativeGeneralFile = null,  bool waitForGeneralData = false, TimeSpan? timeout = null)
        {

            if (alternativeGeneralFile != null && !string.IsNullOrEmpty(alternativeGeneralFile))
            {
                if (_fileSystem.File.Exists(alternativeGeneralFile))
                {
                    Log(LogSeverityEnum.Info, $"Using general file '{alternativeGeneralFile}' before initialization.");
                    var binder = new Binder(_fileSystem, true, true, 5);
                    var list = binder.Load<IEnumerable<MonItemGeneral>>(alternativeGeneralFile)?.ToList();
                    if (list == null) throw new InvalidOperationException($"Cannot load general file '{alternativeGeneralFile}'");

                    var ep = new IPEndPoint(IPAddress.Parse(UriGeneral.DnsSafeHost), UriGeneral.Port);
                    var monGeneral = new MonGeneral() { InstanceIdentifier = instanceIdentifier, Items = list.ToList() };
                        if (!_generalCache.ContainsKey(instanceIdentifier))
                            _generalCache.Add(instanceIdentifier, new MonReceiverGeneralCahedItem(DateTime.Now,ep,monGeneral));
                        else
                        {
                            foreach (var genitem in monGeneral.Items)
                            {
                                if (_generalCache[instanceIdentifier].Data.Items.All(x =>
                                        string.Equals(x.Name, genitem.Name,
                                            StringComparison.InvariantCultureIgnoreCase)))
                                {
                                    _generalCache[instanceIdentifier].Data.Items.Add(genitem);
                                }
                            }
                        }
                }
            }

            if (waitForGeneralData)
                WaitForGeneralData(instanceIdentifier, name, timeout.GetValueOrDefault(Configuration.GeneralReceiverInterval));

            if (_generalCache.TryGetValue(instanceIdentifier, out var item))
            {
                var generalItem = item.Data.Items.FirstOrDefault(x =>
                    string.Equals(x.Name, name, StringComparison.InvariantCultureIgnoreCase));
                if (generalItem == null) throw new InvalidOperationException($"Cannot find descriptor '{name}' in general data cache for instance identifier '{instanceIdentifier}'. Specification is not available yet.");
                return new QMonReceiverHandler(_fileSystem, Configuration.DataCache.Path,instanceIdentifier, generalItem, refreshInterval, _logger);
            }
            throw new InvalidOperationException($"Instance identifier '{instanceIdentifier}' has not been available in general data cache yet.");
        }

        public bool WaitForGeneralData(string instanceIdentifier, string name, TimeSpan timeout)
        {
            Log(LogSeverityEnum.Debug, $"Waiting for general descriptor for instance identifier '{instanceIdentifier}' and channel '{name}' for {timeout.Seconds} sec...");
            var sw = Stopwatch.StartNew();
            while (sw.Elapsed < timeout)
            {
                if (_generalCache.TryGetValue(instanceIdentifier, out var item))
                {
                    var generalItem = item.Data.Items.FirstOrDefault(x =>
                                               string.Equals(x.Name, name, StringComparison.InvariantCultureIgnoreCase));
                    if (generalItem != null)
                    {
                        Log(LogSeverityEnum.Debug, "Descriptor received.");
                        return true;
                    }
                }
                System.Threading.Thread.CurrentThread.Join(100);
            }
            return false;
        }
        #endregion

        #region *** Private overrides ***

        protected override void OnOpen()
        {
            if (WorkerData == null) throw new InvalidOperationException("WorkerData is null");
            var cancel = new CancellationTokenSource();
            var dataReceiverContext = new QMonWorkerContext(((QMonWorkerContext)WorkerData.Context).Client);
            dataReceiverContext.Cancel = cancel.Token;
            _dataReceiverWorker = new ContiguousWorker(async ()=>await OnReceiveData(dataReceiverContext));
            _dataReceiverWorker.Start();
            Log(LogSeverityEnum.Info, $"Data receiver thread on {UriData} started");
        }

        protected override void OnClose()
        {
            _dataReceiverWorker?.Stop();
            _dataReceiverWorker?.Dispose();
            _dataReceiverWorker = null;
            Log(LogSeverityEnum.Info, $"Data receiver thread on {UriData} stopped");
        }

        protected override async Task OnTimerData(QMonWorkerContext context)
        {
            _lastSavedTimestamp = DateTime.Now.ToFileTimeUtc();
            IEnumerable<MonData> itemsToSave = _dataCache.Where(x => x.TimestampFtUtc < _lastSavedTimestamp).ToImmutableArray();
            IEnumerable<string> 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.TimestampFtUtc).ToArray();
                _dataReceivedCallback?.Invoke(batchData);

                foreach (var data in batchData)
                {
                    var sbFileName = new StringBuilder();
                    sbFileName.Append(data.Name).Append("_").Append(data.TimestampFtUtc).Append(".json");
                    var fileName = _fileSystem.Path.Combine(dirName, sbFileName.ToString());
                    await using (var streamWriter = File.CreateText(fileName))
                    {
                        await JsonSerializer.SerializeAsync(streamWriter.BaseStream, data, _jsonSerializerOptions,context.Cancel);
                        await streamWriter.FlushAsync();
                    }
                    Log(LogSeverityEnum.Trace, $"Data receiver created dump data file '{fileName}'");
                }               
                
            }

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

        protected override async Task OnTimerGeneral(QMonWorkerContext context)
        {
            try
            {
                //var remoteEndPoint =
                //    new IPEndPoint(IPAddress.Parse(UriGeneral.DnsSafeHost),UriGeneral.Port);
                var data = await context.Client.ReceiveAsync(context.Cancel);
                var general = (MonGeneral)Binder.LoadFromString(Encoding.UTF8.GetString(data.Buffer), typeof(MonGeneral));
                Log(LogSeverityEnum.Trace,
                    $"[ReceiverGeneral] receive general data from '{data.RemoteEndPoint}' [{general.InstanceIdentifier}] size={data.Buffer.Length} bytes.");

                if (_generalCache.TryGetValue(general.InstanceIdentifier, out var value))
                {
                    foreach (var item in general.Items)
                    {
                        if (!value.Data.Items.Any(x=>string.Equals(x.Name, item.Name, StringComparison.InvariantCultureIgnoreCase)))
                        {
                            value.Data.Items.Add(item);
                            value.Received = DateTime.Now;
                        }
                    }
                }
                else
                {
                    _generalCache.TryAdd(general.InstanceIdentifier,
                        new MonReceiverGeneralCahedItem(DateTime.Now, data.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);
            }
        }

        private async Task OnReceiveData(QMonWorkerContext context)
        {
            while (!context.Cancel.IsCancellationRequested)
            {
                try
                {
                    System.Threading.Thread.CurrentThread.Join(100);
                    Log(LogSeverityEnum.Trace, $"[ReceiverData] Waiting for data on '{UriData}'...");
                    var data = await context.Client.ReceiveAsync(context.Cancel);
                    var monData = JsonSerializer.Deserialize<MonData>(data.Buffer, _jsonSerializerOptions);
                    if (monData == null)
                    {
                        Log(LogSeverityEnum.Warn, $"[ReceiverData] cannot deserialize monitor data from '{data.RemoteEndPoint}' size={data.Buffer.Length} bytes. Skipped.");
                        continue;
                    }
                    _dataCache.Add(monData);
                    Log(LogSeverityEnum.Trace, $"[ReceiverData] receive monitor data from '{data.RemoteEndPoint}' size={data.Buffer.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()
        {
            if (_dataReceiverWorker is { IsRunning: true }) _dataReceiverWorker?.Stop();;
            _dataReceiverWorker?.Dispose();
            _dataReceiverWorker = null;
            base.OnDisposing();
        }


        #endregion


    }
}