bo/Workbench/QMonitor/qmonlib/QMonReceiver.cs

using System.Collections.Concurrent;
using System.Data;
using System.Net;
using System.Net.Sockets;
using System.Text;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Object;

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.ToArray();
        #endregion
        #region *** Fields ***
        private GeneralDataReceivedDelegate _generalReceivedCallback;

            private bool _isReceivingGeneral;
        private IDictionary<string, MonGeneral> _generalCache = new ConcurrentDictionary<string, MonGeneral>();
        #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()
        {
        }

        protected override void OnClose()
        {
        }

        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, general);
                _generalReceivedCallback?.Invoke(general);
            }
            catch (Exception e)
            {
                Log(LogSeverityEnum.Error, $"[ReceiverGeneral] Error receiving general data", e);
            }
            finally
            {
                owner._isReceivingGeneral = false;
            }
        }
        protected override UdpClient CreateUdpClient(Uri targetUri)
        {
            var udpClient = new UdpClient(targetUri.Port);
            return udpClient;
        }

        protected override void OnDisposing()
        {
        }
        #endregion
    }
}