bo/Workbench/QMonitor/qmonlib/QMonReceiverHandler.cs

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<DataCacheItem> _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<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()
        {
            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
    }
}