如何实现类似Switch()仅当新observable发首个值才切换的自定义操作符

核心逻辑说明 #

你需要的Relay操作符和Switch的核心差异在于切换时机，以及新内部流无值终止的处理逻辑，我们可以完全通过原生Rx操作符组合实现，不需要从零自定义Observable。

简洁实现代码 #

using System;
using System.Reactive.Linq;

public static class RelayExtension
{
    public static IObservable<TSource> Relay<TSource>(this IObservable<IObservable<TSource>> source)
    {
        return source
            // 对每个内部流做预处理：仅在首个值到达/流终止时发射结果
            .Select(inner =>
                inner.Take(1)
                    // 只要拿到首个值，就把完整的原始内部流抛出去
                    .Select(_ => inner)
                    // 异常直接透传
                    .Catch<IObservable<TSource>, Exception>(ex => Observable.Throw<IObservable<TSource>>(ex))
                    // 如果内部流没有发任何值就终止，返回空流标记无可用新流
                    .Concat(Observable.Return(Observable.Empty<TSource>()))
                    // 只取第一个结果，避免多余通知
                    .Take(1)
            )
            // 第一层Switch：始终订阅最新的预处理后的内部流
            .Switch()
            // 第二层Switch：切换到实际要消费的内部流
            .Switch();
    }
}

行为匹配验证 #

你可以对照需求确认所有边界场景的处理逻辑：

• 新内部流到达后不会立刻切换：预处理后的流只有在拿到首个值才会抛出原始内部流，第二层Switch此时才会取消旧流订阅，切换到新流，切换前旧流的所有值都会正常下发
• 新内部流无值终止：此时预处理流会返回空Observable，第二层Switch会直接取消旧流订阅，进入无活跃流状态，等待源发出下一个内部流
• 异常、流终止等常规场景都和原生Switch行为保持一致，不会出现额外的状态异常

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