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; namespace qmonlib { 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 GeneralDataReceivedDelegate _generalReceivedCallback; private DataReceivedDelegate _dataReceivedCallback; private bool _isReceivingGeneral; private CancellationTokenSource _dataReceiverCancel; private IDictionary _generalCache = new ConcurrentDictionary(); private IList _dataCache = new List(); private JsonSerializerOptions _jsonSerializerOptions; private DateTime _lastSavedTimestamp; private IFileSystem _fileSystem; #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 *** Private overrides *** protected override void OnOpen() { var _dataReceiverCancel = new CancellationTokenSource(); new 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() { _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 (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 { if (UdpClientData.Available==0) return; var remoteEndPoint = new IPEndPoint(IPAddress.Parse(this.UriData.DnsSafeHost), this.UriData.Port); Log(LogSeverityEnum.Debug, $"[ReceiverData] Waiting for data on '{UriData}'..."); var data = UdpClientData.Receive(ref remoteEndPoint); var monData = JsonSerializer.Deserialize(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 } }