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 { #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; 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(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 } }