bo/Workbench/QMonitor/qmonlib/QMonReceiver.cs

using System.Collections.Concurrent;
using System.IO.Abstractions;
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;
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 List<MonData>();
        private readonly JsonSerializerOptions _jsonSerializerOptions;
        private DateTime _lastSavedTimestamp;
        private readonly IFileSystem _fileSystem;
        private ContiguousWorker? _dataReceiverWorker;
        #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()
        {
            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;
            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 dtFrom = batchData.Min(x => x.Timestamp).ToFileTimeUtc();
                var dtTo = batchData.Max(x => x.Timestamp).ToFileTimeUtc();
                var sbFileName = new StringBuilder();
                sbFileName.Append("mondata_").Append(dtFrom).Append("-").Append(dtTo).Append(".json");
                var fileName = _fileSystem.Path.Combine(dirName, sbFileName.ToString());
                await using (var streamWriter = File.CreateText(fileName))
                {
                    await JsonSerializer.SerializeAsync(streamWriter.BaseStream, batchData, _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.Debug,
                    $"[ReceiverGeneral] receive general data from '{data.RemoteEndPoint}' [{general.InstanceIdentifier}] size={data.Buffer.Length} bytes.");

                _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.Debug, $"[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.Debug, $"[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


    }
}