|
|
@@ -1,42 +1,119 @@
|
|
|
-using Quadarax.Foundation.Core.QMonitor.Interfaces;
|
|
|
+using System.Collections.Concurrent;
|
|
|
+using System.IO.Abstractions;
|
|
|
+using System.Text.Json;
|
|
|
+using Quadarax.Foundation.Core.Collections;
|
|
|
+using Quadarax.Foundation.Core.Json.Extensions;
|
|
|
+using Quadarax.Foundation.Core.Logging;
|
|
|
+using Quadarax.Foundation.Core.QMonitor.Interfaces;
|
|
|
using Quadarax.Foundation.Core.Object;
|
|
|
+using Quadarax.Foundation.Core.Value.Extensions;
|
|
|
|
|
|
namespace Quadarax.Foundation.Core.QMonitor
|
|
|
{
|
|
|
public class QMonReceiverHandler : DisposableObject, IReceiverHandler
|
|
|
{
|
|
|
- #region *** PRoperties ***
|
|
|
+ #region *** Properties ***
|
|
|
|
|
|
public string InstanceIdentifier { get; }
|
|
|
public MonItemGeneral Descriptor { get; }
|
|
|
- public ReceiverHandlerStateEnum State => throw new NotImplementedException();
|
|
|
- public bool IsLiveBroadcast => throw new NotImplementedException();
|
|
|
- public DateTime Timestamp => throw new NotImplementedException();
|
|
|
|
|
|
- public event ReceiverHandlerDataChanged? DataChanged;
|
|
|
+ public ReceiverHandlerStateEnum State
|
|
|
+ {
|
|
|
+ get => _state;
|
|
|
+ private set
|
|
|
+ {
|
|
|
+ _state = value;
|
|
|
+ ReceiverHandlerStateChanged?.Invoke(this, _state);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public bool IsLiveBroadcast {get; private set; }
|
|
|
+ public DateTime? Timestamp { get; private set; }
|
|
|
+ public TimeSpan RefreshInterval { get => _refreshInterval; set => SetRefreshInterval(value); }
|
|
|
+ public MonData[]? CurrentData { get; private set; }
|
|
|
+ public event ReceiverHandlerStateChanged? ReceiverHandlerStateChanged;
|
|
|
+ public event ReceiverHandlerDataChanged? ReceiverHandlerDataChanged;
|
|
|
+
|
|
|
+ #endregion
|
|
|
+
|
|
|
+ #region *** Private fields ***
|
|
|
+
|
|
|
+ private ReceiverHandlerStateEnum _state;
|
|
|
+ private TimeSpan _refreshInterval;
|
|
|
+ private IFileSystem _fileSystem;
|
|
|
+ private string _cachePath;
|
|
|
+ private ILog? _log;
|
|
|
+ private IFileSystemWatcher? _watcher;
|
|
|
+ private Timer _refreshTimer;
|
|
|
+ private readonly LockedList<DataCacheItem> _dataCache = new();
|
|
|
+
|
|
|
+ private bool _isOverlapOnRefreshTimer;
|
|
|
#endregion
|
|
|
|
|
|
#region *** Constructor ***
|
|
|
- internal QMonReceiverHandler(string instanceIdentifier, MonItemGeneral descriptor)
|
|
|
+ internal QMonReceiverHandler(IFileSystem fileSystem, string cachePath, string instanceIdentifier, MonItemGeneral descriptor,TimeSpan refreshInterval, ILogger? logger)
|
|
|
{
|
|
|
+ _fileSystem = fileSystem ?? throw new ArgumentNullException(nameof(fileSystem));
|
|
|
+ if (string.IsNullOrEmpty(cachePath)) throw new ArgumentNullException(nameof(cachePath));
|
|
|
+ _cachePath = cachePath;
|
|
|
+ _log = logger?.GetLogger(nameof(QMonReceiverHandler));
|
|
|
+
|
|
|
if(string.IsNullOrEmpty(instanceIdentifier)) throw new ArgumentNullException(nameof(instanceIdentifier));
|
|
|
|
|
|
InstanceIdentifier = instanceIdentifier;
|
|
|
Descriptor = descriptor ?? throw new ArgumentNullException(nameof(descriptor));
|
|
|
+
|
|
|
+ RefreshFiles();
|
|
|
+ _watcher = _fileSystem.FileSystemWatcher.New(_fileSystem.Path.Combine(_cachePath, InstanceIdentifier),
|
|
|
+ GetFileSearchMask());
|
|
|
+ _watcher.Created += Watcher_Created;
|
|
|
+ _watcher.Deleted += Watcher_Deleted;
|
|
|
+ RefreshInterval = refreshInterval;
|
|
|
+ Timestamp = GetLastAvailableDataTimestamp();
|
|
|
+ _state = ReceiverHandlerStateEnum.Stop;
|
|
|
}
|
|
|
+
|
|
|
#endregion
|
|
|
|
|
|
#region *** Public Operations ***
|
|
|
public DateTime? GetFirstAvailableDataTimestamp()
|
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
+
|
|
|
+ if (_dataCache.Count == 0) return null;
|
|
|
+ var first = _dataCache.Select(x => x.Timestamp).MinBy(x => x);
|
|
|
+ return first;
|
|
|
+
|
|
|
+ }
|
|
|
+ public DateTime? GetLastAvailableDataTimestamp()
|
|
|
+ {
|
|
|
+
|
|
|
+ if (_dataCache.Count == 0) return null;
|
|
|
+ var last = _dataCache.Select(x => x.Timestamp).MinBy(x => x);
|
|
|
+ return last;
|
|
|
+
|
|
|
}
|
|
|
|
|
|
public void Play(bool isLive)
|
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
+ State = ReceiverHandlerStateEnum.Play;
|
|
|
+ if (isLive)
|
|
|
+ {
|
|
|
+ if (_dataCache.Count == 0) return;
|
|
|
+ SetAsCurrent(_dataCache.Last());
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ public void RewindToPreviousChange()
|
|
|
+ {
|
|
|
+ RewindToChange(false);
|
|
|
+ }
|
|
|
+ public void RewindToNextChange()
|
|
|
+ {
|
|
|
+ RewindToChange(true);
|
|
|
}
|
|
|
|
|
|
+
|
|
|
public void Record()
|
|
|
{
|
|
|
throw new NotImplementedException();
|
|
|
@@ -44,19 +121,257 @@ namespace Quadarax.Foundation.Core.QMonitor
|
|
|
|
|
|
public void Rewind(TimeSpan? duration = null, bool isBackward = false)
|
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
+ State = ReceiverHandlerStateEnum.Rewind;
|
|
|
+ if (_dataCache.Count == 0 || _dataCache.Count == 1)
|
|
|
+ {
|
|
|
+ State = ReceiverHandlerStateEnum.Stop;
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (duration == null || duration == TimeSpan.Zero)
|
|
|
+ {
|
|
|
+ if (isBackward)
|
|
|
+ SetAsCurrent(_dataCache.First());
|
|
|
+ else
|
|
|
+ SetAsCurrent(_dataCache.Last());
|
|
|
+
|
|
|
+ State = ReceiverHandlerStateEnum.Stop;
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ var findTimestamp = (Timestamp ?? DateTime.Now);
|
|
|
+ findTimestamp = isBackward ? findTimestamp.Subtract(duration.Value) : findTimestamp.Add(duration.Value);
|
|
|
+ int newIndex = -1;
|
|
|
+ if (isBackward)
|
|
|
+ {
|
|
|
+ for (var i = 0; i < _dataCache.Count; i++)
|
|
|
+ {
|
|
|
+ if (_dataCache[i].Timestamp >= findTimestamp)
|
|
|
+ {
|
|
|
+ newIndex = i;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+
|
|
|
+ for (var i = _dataCache.Count - 1; i >= 0; i--)
|
|
|
+ {
|
|
|
+ if (_dataCache[i].Timestamp <= findTimestamp)
|
|
|
+ {
|
|
|
+ newIndex = i;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ if (newIndex >= 0)
|
|
|
+ {
|
|
|
+ SetAsCurrent(_dataCache[newIndex]);
|
|
|
+ }
|
|
|
+ State = ReceiverHandlerStateEnum.Stop;
|
|
|
}
|
|
|
|
|
|
+
|
|
|
public void Stop()
|
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
+ State = ReceiverHandlerStateEnum.Stop;
|
|
|
}
|
|
|
#endregion
|
|
|
|
|
|
#region *** Private Operations ***
|
|
|
+
|
|
|
+ public int FindNextChangeIndex(int currentIndex, bool isForward)
|
|
|
+ {
|
|
|
+
|
|
|
+ if (currentIndex < 0 || currentIndex >= _dataCache.Count) return -1;
|
|
|
+ var originalIndex = currentIndex;
|
|
|
+ if (isForward)
|
|
|
+ {
|
|
|
+ if (currentIndex >= _dataCache.Count - 1) return -1;
|
|
|
+ currentIndex++;
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ if (currentIndex <= 0) return -1;
|
|
|
+ currentIndex--;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (_dataCache[originalIndex].Data.Diff(_dataCache[currentIndex].Data).Count > 0)
|
|
|
+ return currentIndex;
|
|
|
+
|
|
|
+ return FindNextChangeIndex(currentIndex, isForward);
|
|
|
+ }
|
|
|
+
|
|
|
+ public void RewindToChange(bool isForward)
|
|
|
+ {
|
|
|
+ if (Timestamp == null) return;
|
|
|
+ var timestamp = Timestamp;
|
|
|
+ int currentIndex = -1;
|
|
|
+
|
|
|
+ var currentItem = _dataCache.FirstOrDefault(x => x.Timestamp == timestamp);
|
|
|
+ if (currentItem == null) return;
|
|
|
+ currentIndex = _dataCache.IndexOf(currentItem);
|
|
|
+
|
|
|
+ var nextChangeIndex = FindNextChangeIndex(currentIndex, isForward);
|
|
|
+ if (nextChangeIndex >= 0)
|
|
|
+ SetAsCurrent(_dataCache[nextChangeIndex]);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ private bool SetAsCurrent(DataCacheItem item)
|
|
|
+ {
|
|
|
+ var result = false;
|
|
|
+ using (var fReader = _fileSystem.File.OpenText(item.FileName))
|
|
|
+ {
|
|
|
+
|
|
|
+ var data = JsonSerializer.Deserialize<MonData>(fReader.BaseStream);
|
|
|
+ if (data != null)
|
|
|
+ {
|
|
|
+ ReceiverHandlerDataChanged?.Invoke(this, data);
|
|
|
+ Timestamp = item.Timestamp;
|
|
|
+ CurrentData = new[] { data };
|
|
|
+ result = true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void SetRefreshInterval(TimeSpan refreshInterval)
|
|
|
+ {
|
|
|
+ _refreshInterval = refreshInterval;
|
|
|
+ _refreshTimer?.Dispose();
|
|
|
+ _log?.Log(LogSeverityEnum.Debug, $"[{InstanceIdentifier}/{Descriptor.Name}] Refresh interval was set to {RefreshInterval.ToReadableString()}");
|
|
|
+ _refreshTimer = new Timer(OnRefreshTimer, this, TimeSpan.Zero, refreshInterval);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ private void OnRefreshTimer(object? state)
|
|
|
+ {
|
|
|
+ //TODO: add overlap checking here
|
|
|
+ if (!_isOverlapOnRefreshTimer)
|
|
|
+ {
|
|
|
+ _isOverlapOnRefreshTimer = true;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ if (State == ReceiverHandlerStateEnum.Play)
|
|
|
+ {
|
|
|
+ if (_dataCache.Count == 0) return;
|
|
|
+ if (Timestamp < _dataCache.Last().Timestamp)
|
|
|
+ {
|
|
|
+ var currentItem = _dataCache.FirstOrDefault(x => x.Timestamp == Timestamp);
|
|
|
+ if (currentItem == null) return;
|
|
|
+ var currentIndex = _dataCache.IndexOf(currentItem);
|
|
|
+ var newChangeIndex = FindNextChangeIndex(currentIndex, true);
|
|
|
+ if (newChangeIndex >= 0)
|
|
|
+ SetAsCurrent(_dataCache[newChangeIndex]);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ _log?.Log(LogSeverityEnum.Error,
|
|
|
+ $"[{InstanceIdentifier}/{Descriptor.Name}] Error on refresh timer: {ex.Message}");
|
|
|
+ }
|
|
|
+ finally
|
|
|
+ {
|
|
|
+ _isOverlapOnRefreshTimer = false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ private void RefreshFiles()
|
|
|
+ {
|
|
|
+ var result = new List<Tuple<string, DateTime>>();
|
|
|
+ var dirName = _fileSystem.Path.Combine(_cachePath, InstanceIdentifier);
|
|
|
+ if (!_fileSystem.Directory.Exists(dirName))
|
|
|
+ {
|
|
|
+ _fileSystem.Directory.CreateDirectory(dirName);
|
|
|
+ _log?.Log(LogSeverityEnum.Debug, $"Data receiver created data cache directory '{dirName}'");
|
|
|
+ }
|
|
|
+
|
|
|
+ var files = _fileSystem.Directory.GetFiles(dirName, GetFileSearchMask());
|
|
|
+
|
|
|
+ var start = DateTime.Now;
|
|
|
+
|
|
|
+ //remove missing files (was deleted by retention policy)
|
|
|
+ var cacheItemsToRemove = _dataCache.Where(x => !files.Contains(x.FileName)).ToList();
|
|
|
+ foreach (var item in cacheItemsToRemove)
|
|
|
+ {
|
|
|
+ _dataCache.Remove(item);
|
|
|
+ item.Data.Dispose();
|
|
|
+ }
|
|
|
+
|
|
|
+ var addedCnt = 0;
|
|
|
+ foreach (var file in files)
|
|
|
+ if (MayAddFileToCache(file)) addedCnt++;
|
|
|
+
|
|
|
+ _log?.Log(LogSeverityEnum.Debug,$"[{InstanceIdentifier}/{Descriptor.Name}] Read {addedCnt} files and add to cache @ {DateTime.Now.Subtract(start).TotalMilliseconds} ms.");
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ private bool MayAddFileToCache(string fileName)
|
|
|
+ {
|
|
|
+ if (_dataCache.Any(x=>x.FileName == fileName)) return false;
|
|
|
+ using (var fReader = _fileSystem.File.OpenText(fileName))
|
|
|
+ {
|
|
|
+ var data = JsonSerializer.Deserialize<MonData>(fReader.BaseStream);
|
|
|
+ if (data == null) return false;
|
|
|
+ _dataCache.Add(new DataCacheItem(fileName, data));
|
|
|
+ }
|
|
|
+
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ private string GetFileSearchMask()
|
|
|
+ {
|
|
|
+ return $"{Descriptor.Name}*.json";
|
|
|
+ }
|
|
|
protected override void OnDisposing()
|
|
|
{
|
|
|
- DataChanged = null;
|
|
|
+ ReceiverHandlerStateChanged = null;
|
|
|
+ ReceiverHandlerStateChanged = null;
|
|
|
+ _watcher?.Dispose();
|
|
|
+ _watcher = null;
|
|
|
+ _refreshTimer?.Dispose();
|
|
|
+ }
|
|
|
+ private void Watcher_Deleted(object sender, FileSystemEventArgs e)
|
|
|
+ {
|
|
|
+ if (_dataCache.Any(x => x.FileName == e.FullPath))
|
|
|
+ {
|
|
|
+ var item = _dataCache.First(x => x.FileName == e.FullPath);
|
|
|
+
|
|
|
+ _dataCache.Remove(item);
|
|
|
+ item.Data.Dispose();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void Watcher_Created(object sender, FileSystemEventArgs e)
|
|
|
+ {
|
|
|
+ if (_dataCache.All(x => x.FileName != e.FullPath))
|
|
|
+ {
|
|
|
+ MayAddFileToCache(e.FullPath);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ #endregion
|
|
|
+
|
|
|
+ #region *** Nested Classes ***
|
|
|
+
|
|
|
+ private class DataCacheItem
|
|
|
+ {
|
|
|
+ public DateTime Timestamp { get; set; }
|
|
|
+ public JsonDocument Data { get; set; }
|
|
|
+ public string FileName { get; set; }
|
|
|
+
|
|
|
+ public DataCacheItem(string fileName, MonData data)
|
|
|
+ {
|
|
|
+ Timestamp = data.Timestamp;
|
|
|
+ Data = JsonDocument.Parse(data.Data);
|
|
|
+ FileName = fileName;
|
|
|
+ }
|
|
|
}
|
|
|
#endregion
|
|
|
}
|
