bo/Workbench/QMonitor/qmonlib/QMonReceiver.cs

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<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 => 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<string, MonReceiverGeneralCahedItem> _generalCache = new ConcurrentDictionary<string, MonReceiverGeneralCahedItem>();
        #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
    }
}