using System;

using System.Threading.Tasks;

using System.Reactive.Linq;

​

public static class Program

{

    public static void Main()

    {

        Observable

            .Interval(TimeSpan.FromMilliseconds(200))

            .Select(x => (int)x + 1)

            .Take(10)

            .Do(x => Console.WriteLine($"Produced #{x}"))

            .ExhaustMap(async x => { await Task.Delay(x % 3 == 0 ? 500 : 100); return x; })

            .Do(x => Console.WriteLine($"--Result: {x}"))

            .Wait();

        Console.WriteLine($"Finished");

    }

​

    /// <summary>Invokes an asynchronous function for each element of an observable

    /// sequence, ignoring elements that are emitted before the completion of an

    /// asynchronous function of a preceding element.</summary>

    public static IObservable<TResult> ExhaustMap<TSource, TResult>(

        this IObservable<TSource> source,

        Func<TSource, Task<TResult>> function)

    {

        return source

            .Scan((Task: Task.FromResult<TResult>(default), Id: 0), (previous, item) =>

                !previous.Task.IsCompleted ? previous : (function(item), unchecked(previous.Id + 1)))

            .DistinctUntilChanged()

            .Select(e => e.Task)

            .Concat();

    }

}

​