如何在Rx.Net中实现exhaustMap处理程序？

问题描述

我正在从exhaustMap寻找类似于rxjs运算符的内容，但是RX.NET似乎没有这样的运算符。

我需要实现的是，在源流的每个元素上，我需要启动一个async处理程序，直到完成，我想从源中删除任何元素。处理程序完成后，立即恢复使用元素。

我不希望在每个元素上启动异步处理程序-在处理程序运行时，我想删除源元素。

我还怀疑我需要在这里巧妙地使用defer运算符吗？

谢谢！

解决方法

这是ExhaustMap运算符的实现。可观察到的源投影到IObservable<Task<TResult>>上，其中每个后续任务（如果仍在运行）是前一个任务，否则是与当前项目关联的新任务。然后使用DistinctUntilChanged运算符删除重复出现的同一任务，最后使用Concat运算符将可观察对象展平。

public static IObservable<TResult> ExhaustMap<TSource,TResult>(
    this IObservable<TSource> source,Func<TSource,Task<TResult>> function)
{
    return source
        .Scan(Task.FromResult<TResult>(default),(previousTask,item) =>
        {
            return !previousTask.IsCompleted ? previousTask : HideIdentity(function(item));
        })
        .DistinctUntilChanged()
        .Concat();

    async Task<TResult> HideIdentity(Task<TResult> task) => await task;
}

不能保证function返回的任务是不同的，因此需要HideIdentity局部函数来返回任务的不同包装。

用法示例：

Observable
    .Interval(TimeSpan.FromMilliseconds(200))
    .Select(x => (int)x + 1)
    .Take(10)
    .Do(x => Console.WriteLine($"Input: {x}"))
    .ExhaustMap(async x => { await Task.Delay(x % 3 == 0 ? 500 : 100); return x; })
    .Do(x => Console.WriteLine($"Result: {x}"))
    .Wait();

输出：

Input: 1
Result: 1
Input: 2
Result: 2
Input: 3
Input: 4
Input: 5
Result: 3
Input: 6
Input: 7
Input: 8
Result: 6
Input: 9
Input: 10
Result: 9

c#c#rx.net system.reactive