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.Thread; 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 => _currentTimestampFtUtc == null ? null : DateTime.FromFileTimeUtc(_currentTimestampFtUtc.Value); 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 LoopWorker _refreshTimer; private readonly LockedList _dataCache = new(); private ILogger? _logger; private long? _currentTimestampFtUtc; #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)); _logger = logger; 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; _currentTimestampFtUtc = GetLastAvailableDataTimestamp()?.ToFileTimeUtc(); _state = ReceiverHandlerStateEnum.Stop; } #endregion #region *** Public Operations *** public DateTime? GetFirstAvailableDataTimestamp() { if (_dataCache.Count == 0) return null; var first = _dataCache.Select(x => x.TimestampFtUtc).MinBy(x => x); return DateTime.FromFileTimeUtc(first); } public DateTime? GetLastAvailableDataTimestamp() { if (_dataCache.Count == 0) return null; var last = _dataCache.Select(x => x.TimestampFtUtc).MinBy(x => x); return DateTime.FromFileTimeUtc(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); var findTimestampFtUtc = (isBackward ? findTimestamp.Subtract(duration.Value) : findTimestamp.Add(duration.Value)).ToFileTimeUtc(); int newIndex = -1; if (isBackward) { for (var i = 0; i < _dataCache.Count; i++) { if (_dataCache[i].TimestampFtUtc >= findTimestampFtUtc) { newIndex = i; break; } } } else { for (var i = _dataCache.Count - 1; i >= 0; i--) { if (_dataCache[i].TimestampFtUtc <= findTimestampFtUtc) { 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 (_currentTimestampFtUtc == null) return; var timestamp = _currentTimestampFtUtc; int currentIndex = -1; var currentItem = _dataCache.FirstOrDefault(x => x.TimestampFtUtc == _currentTimestampFtUtc); 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); _currentTimestampFtUtc = item.TimestampFtUtc; CurrentData = new[] { data }; result = true; } } return result; } private void SetRefreshInterval(TimeSpan refreshInterval) { _refreshTimer?.Stop(true); _refreshInterval = refreshInterval; _refreshTimer?.Dispose(); _log?.Log(LogSeverityEnum.Debug, $"[{InstanceIdentifier}/{Descriptor.Name}] Refresh interval was set to {_refreshInterval.ToReadableString()}"); _refreshTimer = new LoopWorker("ReceiverHandler",null,async (_)=>await OnRefreshTimer(this), _logger); _refreshTimer.Start(_refreshInterval); } private async Task OnRefreshTimer(object? state) { //TODO: add overlap checking here if (State == ReceiverHandlerStateEnum.Play) { if (_dataCache.Count == 0) return; if (_currentTimestampFtUtc < _dataCache.Last().TimestampFtUtc) { var currentItem = _dataCache.FirstOrDefault(x => x.TimestampFtUtc == _currentTimestampFtUtc); if (currentItem == null) return; var currentIndex = _dataCache.IndexOf(currentItem); var newChangeIndex = FindNextChangeIndex(currentIndex, true); if (newChangeIndex >= 0) SetAsCurrent(_dataCache[newChangeIndex]); } } } 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 long TimestampFtUtc { get; set; } public JsonDocument Data { get; set; } public string FileName { get; set; } public DataCacheItem(string fileName, MonData data) { TimestampFtUtc = data.TimestampFtUtc; Data = JsonDocument.Parse(data.Data); FileName = fileName; } } #endregion } }