sa/Stars Assistant/Helpers/RxExtensions.cs

using System.Reactive.Concurrency;
using System.Reactive.Linq;
    
namespace StarsAssistant.Helpers;

public static class RxExtensions
{
    /// <summary>
    /// Wrap an observerble to throttle the incoming changes into a given window. Will delay and collect
    /// any events fired until the minimum inactivity period has elapsed without events.
    /// </summary>
    /// <remarks>Curtesy of https://endjin.com/blog/2024/05/observe-file-system-changes-with-rx-dotnet</remarks>
    /// <typeparam name="T">The original type being observed</typeparam>
    /// <param name="src">The source observable to throttle.</param>
    /// <param name="minimumInactivityPeriod">The minimum inactivity period to wait before propagating.</param>
    /// <param name="scheduler">The scheduler to use, defaults to the default scheduler.</param>
    /// <returns>Wrapped observable.</returns>
    public static IObservable<IList<T>> Quiescent<T>(
        this IObservable<T> src, 
        TimeSpan minimumInactivityPeriod, 
        IScheduler? scheduler = null)
    {
        IScheduler realScheduler = scheduler ?? DefaultScheduler.Instance;
        IObservable<int> onoffs =
            from _ in src
            from delta in
                Observable.Return(1, realScheduler)
                    .Concat(Observable.Return(-1, realScheduler)
                        .Delay(minimumInactivityPeriod, realScheduler))
            select delta;

        IObservable<int> outstanding = onoffs.Scan(0, (total, delta) => total + delta);
        IObservable<int> zeroCrossings = outstanding.Where(total => total == 0);

        return src.Buffer(zeroCrossings);
    }

    /// <summary>
    /// Throttle a FileSystemObserver and eliminate all duplicates.
    /// </summary>
    /// <remarks>
    /// Curtesy of https://endjin.com/blog/2024/05/observe-file-system-changes-with-rx-dotnet
    /// </remarks>
    /// <param name="watcher">An observer created with <c>ObserveFileSystem</c></param>
    /// <param name="inactivitySeconds">Throttling window, defaults to 1 second.</param>
    /// <param name="scheduler">Scheduler to user for throttling, default is set by <c>Quiescent</c></param>
    /// <returns>A throttled observer with duplicate elimination.</returns>
    public static IObservable<IEnumerable<FileSystemEventArgs>> ThrottleAndDistinct (
        this IObservable<FileSystemEventArgs> watcher, 
        int inactivitySeconds = 1,
        IScheduler? scheduler = null)
    {
        return watcher
                .Quiescent(TimeSpan.FromSeconds(inactivitySeconds), scheduler)
                .Select(changes => changes.DistinctBy(x => (x.ChangeType, x.FullPath)));
    }
}