使用 Switch 语句时如何处理内部可观察对象

Question

我有一个嵌套的 IObservable，我正在使用来自 Rx 的 switch 语句，它帮助我处理以前的序列。但是如果我想手动处理怎么办？处理外部序列不是一个选项。

_performSearchSubject
.Select(_ => return PerformQuery())
            .Switch()
            .Subscribe(HandleResponseStream, HandleError);

执行查询 returns IObservable<Result>;

Answer 1

过了一段时间我自己找到了...所以答案是：

你可以使用TakeUntil(IObservable<TOther>)，然后传入一个新的主题，你可以在你想取消之前的流时调用它。我认为它类似于底层流中的 Switch() 语句。

最终代码看起来像这样：

Subject<Unit> _cancellationObservable = new Subject<Unit>();

_performSearchSubject
.Select(_ => {
                return PerformQuery().TakeUntil(_cancellationObservable);
              })
            .Switch()
            .Subscribe(HandleResponseStream, HandleError);

每当我想取消它时，我都会打电话给这个人：

_cancellationObservable.OnNext(Unit.Default);

Answer 2

'PerformQuery' 返回的 Observable 流将在以下任一情况下自动处理：

您的 'Switch' 语句由于新的“_performSearchSubject”而执行。即，先前的流将由 'Switch' 运算符清理。或者
您的整体查询已处理。

您应该存储您的 .Subscribe 返回的一次性用品，并用它来清理您的订阅。您没有必要使用内部主题。

你真的尝试过你原来的解决方案吗？下面是一些 Linqpad 代码来证明这一点：

var inner = Observable.Create<string>((o) =>
{
    o.OnNext("First item of new inner");
    return Disposable.Create(() => "Inner Disposed".Dump());
});

var outer = Observable.Timer(TimeSpan.MinValue, TimeSpan.FromSeconds(10))
        .Select(_ => inner)
        .Switch()
        .Subscribe(output => output.Dump());


Console.ReadLine();

outer.Dispose();

您会注意到，无论您何时按下一个键并输入（移过 Console.ReadLine），您的内部可观察流都将被处理掉。同样，您的 Switch 语句将在每次触发 Timer 时清除之前的所有流。

如果你希望基于不相关的流来控制它，你仍然可以使用 'TakeUntil'，但是，如果可以的话，最好只使用一次性返回。

使用 Switch 语句时如何处理内部可观察对象

How do I dispose inner observable while using Switch statement

.net

c#

system.reactive