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 { #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 _generalCache = new ConcurrentDictionary(); private readonly IList _dataCache = new List(); 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(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 } }