using System.Collections.Concurrent; using System.Net; using System.Net.Sockets; using System.Text; using System.Text.Json; using Quadarax.Foundation.Core.Logging; namespace qmonlib { public delegate void GeneralDataReceivedDelegate(MonGeneral general); 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 => null; public MonGeneral[] AvailableGeneral => _generalCache.Values.Select(x=>x.Data).ToArray(); #endregion #region *** Fields *** private GeneralDataReceivedDelegate _generalReceivedCallback; private bool _isReceivingGeneral; private CancellationTokenSource _dataReceiverCancel; private IDictionary _generalCache = new ConcurrentDictionary(); #endregion #region *** Constructors *** public QMonReceiver(QMonReceiverConfiguration configuration, ILogger logger, GeneralDataReceivedDelegate generalDataReceivedCallback = null) : base(configuration, logger) { _generalReceivedCallback = generalDataReceivedCallback; } #endregion #region *** Private overrides *** protected override void OnOpen() { var _dataReceiverCancel = new CancellationTokenSource(); new Thread(() => { try { OnThreadData(_dataReceiverCancel.Token); } catch (OperationCanceledException) { Log(LogSeverityEnum.Info, $"Data receiver thread on {UriData} unexpected stopped"); } }).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) { throw new NotSupportedException("Timer is not allowed in receiver."); } 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}'."); } try { if (UdpClientData.Available==0) return; var remoteEndPoint = new IPEndPoint(IPAddress.Parse(this.UriData.DnsSafeHost), this.UriData.Port); var data = UdpClientData.Receive(ref remoteEndPoint); var jsonData = JsonDocument.Parse(Encoding.UTF8.GetString(data)); 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 } }