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 *** public string InstanceIdentifier { get; } public MonItemGeneral Descriptor { get; } 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 _dataCache = new(); private bool _isOverlapOnRefreshTimer; #endregion #region *** Constructor *** 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() { 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) { 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(); } public void Rewind(TimeSpan? duration = null, bool isBackward = false) { 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() { 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(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>(); 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(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() { 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 } }