Reactive Extensions - Switch, but wait until previous observable completes instead cancelling

Question

I have and async Load method that is triggered by a observable subscription.

I want to prevent concurrent calls of LoadAsync using following strategy:

I'm almost sure this strategy has a name, but I can't recall it.

1. LoadAsync won't be called if previous calls has not finished yet.

   • LoadAsync(3) has started only after LoadAsync(1) has finished

2. LoadAsync is called only for the most recent value

   • LoadAsync(2) has never been called

---

I currently use this switchmap equivalent, but it cancels the LoadAsync methods in favor of new calls.

Observable<int> sourceObservable..
sourceObservable
   .Select(value => Observable.FromAsync(() => LoadAsync(value)))
   .Switch() 

async Task<int> LoadAsync(int value)
{
    await Task.Delay(1000);
    return value;
}

in javascript it would be:

source$.pipe(switchMap(Load))

function load(id): Observable<int> {
    return delay(1000).pipe(map(() => id));
}

EDIT: I wonder why nobody asked for this flattening strategy yet. I thought it's pretty common requirement.

This is how CI/CD pipelines often works, for example. If there are too many pushes to a branch and you don't have the capacity to build them all, but you want to ensure that the latest commit will be built

Answer 1

In other words, you want to take one child observable from the 2nd-level observable, get all results and wait for that to finish, then repeat.

This will work in C#:

public static IObservable<T> SingleSwitch<T, U>(this IObservable<U> source, Func<U, IObservable<T>> selector)
{
    return source.Publish(_source => _source
        .Take(1)
        .SelectMany(e => selector(e))
        .Repeat()
    );
}

Please note that you have a bug in your Load. You'll need to add await for it to the delay properly work.

Sample code:

Observable.Interval(TimeSpan.FromSeconds(.75))
    .SingleSwitch(e => Observable.FromAsync (async () => {
        await Task.Delay(1000);
        return e;
    }))
    .Dump();

Output is even numbers, odd ones get blocked because the evens are running.

Answer 2

MergeBounded(1,1) mentioned by @TheodorZoulias works, but I it seems a little bit too complex.

I have written a simpler alternative, but I'm looking for most appropriate name

/// <summary>
/// Concatenates most recent inner observable sequence when previous completes.
/// Similar to Concat, but it ignores out of date inner observable sequences.
/// Similar to Exhaust, but it preserves latest inner observable.
/// </summary>
public static IObservable<T> ConcatExhaust<T>(this IObservable<IObservable<T>> source)
{
    return Observable.Defer(() =>
    {
        IObservable<T> latest = default;
        return source
            .Select(inner =>
            {
                latest = inner;
                return Observable.Defer(() => latest == inner ? inner : Observable.Empty<T>());
            })
            .ConcatExhaust();
    });
}

The following test:

var source = Observable.Interval(TimeSpan.FromMilliseconds(300))
    .Take(5)
    .Do(val => Console.WriteLine($"Source: {val}"));


source.Select(val => Observable.FromAsync(() => LoadAsync(val)))
     .ConcatExhaust()
     .Wait();

returns:

Source: 0
Load Started: 0
Source: 1
Source: 2
Source: 3
Value finished: 0
Load Started: 3
Source: 4
Value finished: 3
Load Started: 4
Value finished: 4