|
|
@@ -1,10 +1,9 @@
|
|
|
using System.Collections.Concurrent;
|
|
|
-using System.Data;
|
|
|
using System.Net;
|
|
|
using System.Net.Sockets;
|
|
|
using System.Text;
|
|
|
+using System.Text.Json;
|
|
|
using Quadarax.Foundation.Core.Logging;
|
|
|
-using Quadarax.Foundation.Core.Object;
|
|
|
|
|
|
namespace qmonlib
|
|
|
{
|
|
|
@@ -16,13 +15,13 @@ namespace qmonlib
|
|
|
protected override Uri UriData => Configuration.SourceUriData;
|
|
|
protected override TimeSpan IntervalGeneral => Configuration.GeneralReceiverInterval;
|
|
|
protected override TimeSpan? IntervalData => null;
|
|
|
- public MonGeneral[] AvailableGeneral => _generalCache.Values.ToArray();
|
|
|
+ public MonGeneral[] AvailableGeneral => _generalCache.Values.Select(x=>x.Data).ToArray();
|
|
|
#endregion
|
|
|
#region *** Fields ***
|
|
|
private GeneralDataReceivedDelegate _generalReceivedCallback;
|
|
|
-
|
|
|
- private bool _isReceivingGeneral;
|
|
|
- private IDictionary<string, MonGeneral> _generalCache = new ConcurrentDictionary<string, MonGeneral>();
|
|
|
+ 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)
|
|
|
@@ -36,10 +35,25 @@ namespace qmonlib
|
|
|
|
|
|
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)
|
|
|
@@ -61,7 +75,7 @@ namespace qmonlib
|
|
|
Log(LogSeverityEnum.Debug,
|
|
|
$"[ReceiverGeneral] receive general data from '{remoteEndPoint}' [{general.InstanceIdentifier}] size={data.Length} bytes.");
|
|
|
|
|
|
- _generalCache.TryAdd(general.InstanceIdentifier, general);
|
|
|
+ _generalCache.TryAdd(general.InstanceIdentifier, new MonReceiverGeneralCahedItem(DateTime.Now, remoteEndPoint, general));
|
|
|
_generalReceivedCallback?.Invoke(general);
|
|
|
}
|
|
|
catch (Exception e)
|
|
|
@@ -73,6 +87,29 @@ namespace qmonlib
|
|
|
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);
|
|
|
@@ -81,6 +118,10 @@ namespace qmonlib
|
|
|
|
|
|
protected override void OnDisposing()
|
|
|
{
|
|
|
+ _dataReceiverCancel?.Cancel();
|
|
|
+ _dataReceiverCancel?.Dispose();
|
|
|
+ _dataReceiverCancel = null;
|
|
|
+ base.OnDisposing();
|
|
|
}
|
|
|
#endregion
|
|
|
}
|