c# - TestScheduler 创建的 ColdObserver 在 OnCompleted 时不取消订阅？

Question

我想检查我创建的 IObservable 是否尊重“完成后，我将取消订阅你” 的礼貌。乍一看，我的代码似乎有问题。但是删除我的代码，只使用 TestScheduler 提供的 Observable 和 Observer，看起来“取消订阅”永远不会发生:

using Microsoft.Reactive.Testing;
using System.Reactive;
...
var ts = new TestScheduler();
var ob = ts.CreateObserver<int>();
var xs = ts.CreateColdObservable<int>(
    new Recorded<Notification<int>>(1, Notification.CreateOnCompleted<int>())
    );

xs.Subscribe(ob);
ts.AdvanceTo(2);
Assert.Equal(1, xs.Subscriptions.Single().Unsubscribe); //<-- Xunit no like

我最初怀疑观察者，但我在找到的代码变体上尝试了 here ，并且它有效，所以现在我认为 ColdObservable 上的 Subscribe 的实现行为不正常。

Answer 1

不存在这样的礼貌。 RX design guideines在第 4.3 节中建议您可以:

Assume resources are cleaned up after an OnError or OnCompleted message.

在 4.4 节中说你可以:

Assume a best effort to stop all outstanding work on Unsubscribe

这些指南(“礼貌”)讨论了运营商尽快释放自己的资源以及已获得的资源。

在您的代码中，您没有针对这两种情况进行测试。 Unsubscribe的目的ITestableObservable 上的属性(property)是报告观察者取出的订阅何时被显式处置，而不是何时发生内部清理 - 但您没有存储此句柄以便能够处置它:

xs.Subscribe(ob); /* return of handle ignored here */

所以你试图断言你处理了你扔掉的订阅，而不是你订阅的可观察对象清理了它可能已经取出的任何订阅和资源。

如果想看4.3/4.4及时资源清理的效果，写个这样的扩展方法:

public static IObservable<T> SpyResourceCleanUp<T>(
    this IObservable<T> source, IScheduler scheduler)
{
    return Observable.Create<T>(obs =>
    {
        var subscription = source.Subscribe(obs);
        return new CompositeDisposable(
            subscription,
            Disposable.Create(() => Console.WriteLine(
                "Clean up performed at " + scheduler.Now.Ticks)));
    });
}

并替换你的行:

xs.Subscribe(ob);

与

xs.SpyResourceCleanUp(ts).Subscribe(ob);

(编辑一些评论)

在您的测试中，我看到了立即的资源清理，正如我所期望的那样。通过此更改，您的测试现在将通过，因为 SpyResourceCleanUp一旦 OnCompletes() 本身遵守指南的 4.3，就会取消订阅它的父级 (xs)。

这里可能不太明显的是 Observable.Create处理调用 Dispose()返回的方法 IDisposable一旦订阅被处理或 OnComplete()或 OnError()已被要求观察员。这是如何Create帮助您实现第 4.3 节，以及为什么测试通过更改后的代码。

在幕后，订阅 AnonymousObservable<T> : ObservableBase<T>由 Create 返回被 AutoDetachObserver as you can see here 包裹着.

即Disposable你从Observable.Create回来不是调用者得到的 - 他们得到一个包装版本，它将在流终止或取消时调用你的 Dispose()。