如何在Rx.NET和UniRx中实现SelectManyLatest操作符

SelectManyLatest 对应其他Rx框架中的 FlatMapLatest，功能与 SelectMany 类似，核心特性为 当有新的上游事件发射时，会自动终止上一个内部序列的订阅，仅处理最新内部序列产生的事件，非常适合搜索联想、重复请求中断这类需要自动废弃旧任务的场景。

核心实现逻辑 #

• 订阅上游可观察序列，同时维护一个可序列化销毁的内部序列订阅容器
• 每次上游下发新元素时：
  • 立即销毁前一个内部序列的订阅，避免旧序列的事件继续下发
  • 调用传入的选择器生成新的内部可观察序列
  • 订阅新的内部序列，将内部序列的Next、Error事件直接转发给下游，内部序列的Completed事件不直接触发下游完成，等待上游整体完成后再触发下游完成
• 上游抛出错误时，直接将错误传递给下游并终止整个序列
• 上游发送完成事件时，等待当前正在执行的内部序列完成后，再触发下游的完成事件
• 下游主动取消订阅时，同时销毁上游订阅和当前内部序列的订阅

代码实现（基于Rx.NET） #

using System;
using System.Reactive;
using System.Reactive.Disposables;
using System.Reactive.Linq;

public static class ObservableExtensions
{
    public static IObservable<TResult> SelectManyLatest<TSource, TResult>(
        this IObservable<TSource> source,
        Func<TSource, IObservable<TResult>> selector)
    {
        if (source == null) throw new ArgumentNullException(nameof(source));
        if (selector == null) throw new ArgumentNullException(nameof(selector));

        return Observable.Create<TResult>(observer =>
        {
            // 串行管理内部序列的订阅，每次赋值新的Disposable时自动销毁旧的
            var innerSubscription = new SerialDisposable();
            var gate = new object();
            bool isUpstreamCompleted = false;

            var upstreamSubscription = source.Subscribe(
                onNext: value =>
                {
                    lock (gate)
                    {
                        // 销毁前一个内部序列的订阅
                        innerSubscription.Disposable = Disposable.Empty;
                        IObservable<TResult> innerSequence;
                        try
                        {
                            innerSequence = selector(value);
                        }
                        catch (Exception ex)
                        {
                            observer.OnError(ex);
                            return;
                        }

                        var innerObserver = Observer.Create<TResult>(
                            onNext: observer.OnNext,
                            onError: observer.OnError,
                            onCompleted: () =>
                            {
                                lock (gate)
                                {
                                    innerSubscription.Disposable = Disposable.Empty;
                                    // 上游已经完成，且最后一个内部序列也完成时，触发下游完成
                                    if (isUpstreamCompleted)
                                    {
                                        observer.OnCompleted();
                                    }
                                }
                            }
                        );

                        innerSubscription.Disposable = innerSequence.Subscribe(innerObserver);
                    }
                },
                onError: observer.OnError,
                onCompleted: () =>
                {
                    lock (gate)
                    {
                        isUpstreamCompleted = true;
                        // 没有正在执行的内部序列时，直接触发下游完成
                        if (innerSubscription.Disposable == Disposable.Empty)
                        {
                            observer.OnCompleted();
                        }
                    }
                }
            );

            return new CompositeDisposable(upstreamSubscription, innerSubscription);
        });
    }
}

常用重载扩展 #

你可以基于上述基础实现，扩展出更多适用场景的重载：

• 带索引的选择器重载：在上游订阅中维护计数变量，每次下发元素时将索引传入选择器即可
• 带结果选择器的重载：在内部序列的OnNext回调中，将上游原始值和内部序列返回值传入结果选择器，转换后再发送给下游
• 支持Task返回值的重载：将传入的Task异步函数通过ToObservable()转换为可观察序列，复用现有逻辑即可

使用示例 #

以下示例实现了搜索框输入时自动触发查询，且每次新输入都会取消上一次未完成的查询请求，避免旧请求的结果覆盖最新输入的查询结果：

// 监听搜索框输入事件流，防抖300ms避免频繁触发请求
var inputStream = Observable.FromEventPattern<TextChangedEventArgs>(searchBox, nameof(TextBox.TextChanged))
    .Select(_ => searchBox.Text)
    .Throttle(TimeSpan.FromMilliseconds(300))
    .DistinctUntilChanged();

// 每次输入变化触发查询，自动取消上一次未完成的请求
var searchResultStream = inputStream.SelectManyLatest(keyword => SearchApi.GetSearchResultAsync(keyword).ToObservable());

// 订阅查询结果更新UI
searchResultStream
    .ObserveOn(SynchronizationContext.Current)
    .Subscribe(result => 
    {
        resultList.ItemsSource = result;
    }, ex => 
    {
        // 处理错误
        ShowErrorTips(ex.Message);
    });

内容的提问来源于stack exchange，提问作者Dmitry iLinikh