finalize Receiver

Workbench/QMonitor/qmonlib/Dependencies/Json/ListConverter.cs

@@ -1,6 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Text.Json;
+using System.Text.Json;
 using System.Text.Json.Serialization;
 
 namespace Quadarax.Foundation.Core.Json

Workbench/QMonitor/qmonlib/MonData.cs

@@ -0,0 +1,18 @@
+using System.Text.Json.Serialization;
+
+namespace qmonlib
+{
+    public class MonData
+    {
+        [JsonPropertyName("iid")]
+        public string InstanceIdentifier { get; set; }
+
+        [JsonPropertyName("nam")]
+        public string Name { get; set; }
+        [JsonPropertyName("tms")]
+        public DateTime Timestamp { get; set; }
+        [JsonPropertyName("dta")]
+        public string Data { get; set; }
+
+    }
+}

Workbench/QMonitor/qmonlib/MonGeneral.cs

@@ -1,8 +1,11 @@
-namespace qmonlib
+using System.Text.Json.Serialization;
+
+namespace qmonlib
 {
     [Serializable]
     public class MonGeneral
     {
+        [JsonPropertyName("iid")]
         public string InstanceIdentifier { get; set; }
         public List<MonItemGeneral> Items { get; set; } = new List<MonItemGeneral>();
     }

Workbench/QMonitor/qmonlib/Program.cs

@@ -1,7 +1,9 @@
 // See https://aka.ms/new-console-template for more information
 
 
+using System.IO.Abstractions;
 using qmonlib;
+using qmonlib.Test;
 using Quadarax.Foundation.Core.Logging;
 
 var logger = new ConsoleLogger();
@@ -13,9 +15,31 @@ new Thread(() =>
     OnStartReceiver(cancel.Token);
 
 }).Start();
+Console.WriteLine("Waiting for receiver starts...");
+Thread.Sleep(5000);
 using (var cli = new QMonClient("TST", QMonClientConfiuration.CreateDefault(), logger))
 {
 
+    cli.Notify(new TestItem()
+    {
+        Age = 4,
+        FirstName = "Adam",
+        LastName = "Votruba"
+    });
+    cli.Notify(new TestItem()
+    {
+        Age = 47,
+        FirstName = "Dalibor",
+        LastName = "Votruba"
+    });
+    cli.Notify(new TestItem()
+    {
+        Age = 43,
+        FirstName = "Dita",
+        LastName = "Rychetská"
+    });
+
+
     Console.WriteLine("Press any key to close...");
     Console.ReadKey();
     cancel.Cancel();
@@ -26,9 +50,14 @@ void OnGeneralReceived(MonGeneral general)
     Console.WriteLine($"General received: {general.InstanceIdentifier} items count {general.Items.Count}");
 }
 
+void OnDataReceived(MonData[] data)
+{
+    Console.WriteLine($"Data received: Count={data.Length}");
+}
+
 void OnStartReceiver(CancellationToken cancelToken)
 {
-    using (var rcv = new QMonReceiver(QMonReceiverConfiguration.CreateDefault(), logger, OnGeneralReceived))
+    using (var rcv = new QMonReceiver( new FileSystem(), QMonReceiverConfiguration.CreateDefault(), logger, OnGeneralReceived, OnDataReceived))
     {
         while (!cancelToken.IsCancellationRequested)
             Thread.Sleep(100);

Workbench/QMonitor/qmonlib/QMonClient.cs

@@ -24,7 +24,7 @@ namespace qmonlib
 
         #region *** Fields ***
         private IDictionary<string, MonItemGeneral> _generalCache = new Dictionary<string, MonItemGeneral>();
-        
+        private IList<MonData> _dataCache = new List<MonData>();
         #endregion
 
         #region *** Constructors ***
@@ -84,6 +84,21 @@ namespace qmonlib
         #endregion
 
         #region *** Public Operations ***
+        public void Notify(object? monitoredItem)
+        {
+            if (monitoredItem == null) return;
+            var fullName = monitoredItem.GetType().FullName;
+            if (fullName != null && !_generalCache.ContainsKey(fullName)) throw new NotSupportedException($"Type {monitoredItem.GetType().FullName} is not flagged by MonitoredClassAttribute as monitored");
+            
+            var data = new MonData()
+            {
+                InstanceIdentifier = InstanceIdentifier,
+                Name = fullName,
+                Timestamp = DateTime.Now,
+                Data = Binder.SaveToString(monitoredItem)
+            };
+            _dataCache.Add(data);
+        }
         #endregion
 
         #region *** Protected Operations ***
@@ -95,7 +110,30 @@ namespace qmonlib
 
         protected override void OnTimerData(object? state)
         {
-            
+            if (state == null) throw new ArgumentNullException(nameof(state));
+            var owner = (QMonClient)state;
+            if (_dataCache.Count == 0) return;
+            try
+            {
+                var start = DateTime.Now;
+                var itmCnt = 0;
+                var declaredSize = 0;
+                var sentSize = 0;
+                var dataToSent = owner._dataCache.ToArray();
+                foreach (var data in dataToSent)
+                {
+                    var dataOut = Serialize(data);
+                    declaredSize += dataOut.Length;
+                    sentSize += owner.UdpClientGeneral.Send(dataOut);
+                    _dataCache.Remove(data);
+                    itmCnt++;
+                }
+                Log(LogSeverityEnum.Trace, $"[{InstanceIdentifier}] Sent data blocks = {itmCnt} of delared {declaredSize} bytes/sent {sentSize} bytes @ {(DateTime.Now.Subtract(start).TotalMilliseconds)} ms.");
+            }
+            catch(Exception ex)
+            {
+                Log(LogSeverityEnum.Error, $"[{InstanceIdentifier}] Error sending general data", ex);
+            }
         }
 
         protected override void OnTimerGeneral(object? state)
@@ -121,7 +159,7 @@ namespace qmonlib
             }
         }
 
-        private byte[] Serialize(MonGeneral data)
+        private byte[] Serialize(object data)
         {
             if (data == null) throw new ArgumentNullException(nameof(data));
             var json = Binder.SaveToString(data);
@@ -137,11 +175,12 @@ namespace qmonlib
 
         protected override void OnOpen()
         {
+            Log(LogSeverityEnum.Info, $"Started.");
         }
 
         protected override void OnClose()
         {
-            
+            Log(LogSeverityEnum.Info, $"Closed.");
         }
         #endregion
 

Workbench/QMonitor/qmonlib/QMonHostBase.cs

@@ -60,11 +60,15 @@ namespace qmonlib
 
             _udpClientGeneral = CreateUdpClient(UriGeneral);
             _udpClientData = CreateUdpClient(UriData);
-
+            
             _timerGeneral = new Timer(OnTimerGeneral, this, IntervalGeneral, IntervalGeneral);
+            Log(LogSeverityEnum.Debug, $"General timer started with interval '{IntervalGeneral}'");
             if (IntervalData.HasValue)
+            {
                 _timerData = new Timer(OnTimerData, this, IntervalData.Value, IntervalData.Value);
-            
+                Log(LogSeverityEnum.Debug, $"Data timer started with interval '{IntervalData}'");
+            }
+
             OnOpen();
         }
         protected void Close()

Workbench/QMonitor/qmonlib/QMonReceiver.cs

@@ -1,32 +1,60 @@
 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;
 
 namespace qmonlib
 {
     public delegate void GeneralDataReceivedDelegate(MonGeneral general);
+    public delegate void DataReceivedDelegate(MonData[] data);
     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;
+        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<string, MonReceiverGeneralCahedItem> _generalCache = new ConcurrentDictionary<string, MonReceiverGeneralCahedItem>();
+        private IList<MonData> _dataCache = new List<MonData>();
+        private JsonSerializerOptions _jsonSerializerOptions;
+        private DateTime _lastSavedTimestamp;
+        private IFileSystem _fileSystem;
         #endregion
         #region *** Constructors ***
-        public QMonReceiver(QMonReceiverConfiguration configuration, ILogger logger, GeneralDataReceivedDelegate generalDataReceivedCallback = null) : base(configuration, logger)
+        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
         
@@ -40,12 +68,15 @@ namespace qmonlib
             {
                 try
                 {
-                    OnThreadData(_dataReceiverCancel.Token);
+                    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");
         }
@@ -58,7 +89,37 @@ namespace qmonlib
 
         protected override void OnTimerData(object? state)
         {
-            throw new NotSupportedException("Timer is not allowed in receiver.");
+            _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)
@@ -101,7 +162,13 @@ namespace qmonlib
                 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));
+                var monData = JsonSerializer.Deserialize<MonData>(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)
@@ -123,6 +190,10 @@ namespace qmonlib
             _dataReceiverCancel = null;
             base.OnDisposing();
         }
+
+        
         #endregion
+
+        
     }
 }