Observable.Do处理空序列抛出InvalidOperationException的原因及测试解决方法

1. 为何会抛出“Sequence contains no elements”异常？ #

异常并非来自Observable.Do，而是**await IObservable时的默认行为导致**：
当你await一个IObservable<T>时，Reactive Extensions内部会通过AsyncSubject<T>来订阅目标序列。AsyncSubject<T>的特性是仅保存序列的最后一个值，当序列完成时将该值返回给订阅者。如果序列是空的（没有发射任何值就触发完成事件），AsyncSubject<T>没有值可以返回，调用其GetResult()方法时就会抛出InvalidOperationException，也就是你看到的错误。

Observable.Do只是拦截序列事件，不会改变序列的本质，空序列经过Do处理后仍然是空序列，所以问题的核心在于await空IObservable的行为。

2. 单元测试中的解决方法 #

针对空序列的测试需求，提供几种可行方案：

方案1：使用Wait()替代await #

Wait()方法仅等待序列完成，空序列正常完成时不会抛出异常（只有序列出错时才会抛），刚好符合你的测试场景：

[Fact]
public void TestEmpty() {
    IObservable<int> observable = Observable.Empty<int>()
        .Delay(System.TimeSpan.FromSeconds(1.2));

    observable.Do(
        onNext: _ => AssertEx.Fail()
    ).Wait();
}

方案2：用ToTask()指定默认值 #

如果必须使用await，可以通过ToTask()方法指定空序列的默认返回值，避免无值可返回的异常：

[Fact]
public async Task TestEmpty() {
    IObservable<int> observable = Observable.Empty<int>()
        .Delay(System.TimeSpan.FromSeconds(1.2));

    await observable.Do(
        onNext: _ => AssertEx.Fail()
    ).ToTask(defaultValue: default(int));
}

注意：这里要把测试方法的返回类型从async void改为async Task，这是XUnit中异步测试的规范写法（async void可能导致测试框架无法正确捕获异常或等待完成）。

方案3：手动订阅并验证完成事件 #

通过TaskCompletionSource监听序列的完成事件，确保onNext未被触发：

[Fact]
public async Task TestEmpty() {
    IObservable<int> observable = Observable.Empty<int>()
        .Delay(System.TimeSpan.FromSeconds(1.2));

    var tcs = new TaskCompletionSource<bool>();

    observable.Do(
        onNext: _ => AssertEx.Fail(),
        onCompleted: () => tcs.SetResult(true)
    ).Subscribe();

    await tcs.Task;
}

方案4：使用Rx官方测试库（更专业的Rx测试方式） #

借助ReactiveTest和TestScheduler可以精准控制时间并验证序列行为，适合复杂的Rx测试场景：
首先安装NuGet包System.Reactive.Testing，然后编写测试：

[Fact]
public void TestEmptyWithTestScheduler() {
    var scheduler = new TestScheduler();
    
    // 模拟延迟的空序列
    var observable = scheduler.CreateColdObservable<int>(
        ReactiveTest.OnCompleted<int>(TimeSpan.FromSeconds(1.2).Ticks)
    );

    var observer = scheduler.CreateObserver<int>();
    
    observable.Do(
        onNext: _ => AssertEx.Fail()
    ).Subscribe(observer);

    scheduler.Start();

    // 验证序列没有发射任何值，且已完成
    Assert.Empty(observer.Messages);
    Assert.True(observer.IsCompleted);
}

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