using System;

using System.Reactive.Subjects;

using System.Threading.Tasks;

using System.Threading;

using System.Reactive;

using System.Reactive.Linq;

var source = new Subject<int>();

​

var loader = new ReactiveLoader<int>(source, Load);

loader.IsLoadingObservable.Subscribe(val => Console.WriteLine($"Is Loading: {val}"));

​

source.OnNext(2);

await Task.Delay(20);

source.OnNext(4);

await Task.Delay(200);

source.OnNext(5);

await Task.Delay(20);

​

source.OnCompleted();

​

Console.WriteLine("Program Finished");

await Task.Delay(100);

​

static async Task Load(int id, CancellationToken ct) 

{

    Console.WriteLine($"Loading started: {id}");

    try

    {

        await Task.Delay(100, ct);

        Console.WriteLine($"Loading finished: {id}");

    }

    catch (TaskCanceledException) 

    {

        Console.WriteLine($"Loading cancelled: {id}");

    }

​

}

​

​

public class ReactiveLoader<T> : IDisposable

{

    private readonly BehaviorSubject<bool> _isLoading = new(false);

    private readonly Subject<Unit> _completes = new();

    private readonly Subject<T> _reloads = new ();

    private readonly IDisposable _subscription;

​

    public bool IsLoading => _isLoading.Value;

    public IObservable<bool> IsLoadingObservable => _isLoading.Skip(1).DistinctUntilChanged(); //Not nice

​

    public ReactiveLoader(IObservable<T> observable, Func<T, CancellationToken, Task> handler)

    {           

        _subscription = observable

            .Finally(() => //Not nice

            {

                 _completes.OnNext(Unit.Default);

            })

            .Merge(_reloads)

            .Do(_ => _isLoading.OnNext(true))

            .Select(value => Observable.FromAsync(cancellationToken => handler(value, cancellationToken)))

            .Switch()

            .Do(_ => _isLoading.OnNext(false))

            .TakeUntil(_completes) //cancels loading when observable completes

            .Subscribe();

    }

​

    public void Dispose()

    {

        _completes.OnNext(Unit.Default);

        _subscription.Dispose();

    }

}

​