c# - 在 Observable Items 生成时处理它们

Question

我有一个 IObservable 生成一次性元素，并且它会在其生命周期内生成无限数量的元素。因此，我想在每次生成新项目时处理最后一项，所以 Using运营商不会为此工作。是否有不同的 Rx.NET 运算符可以完成此功能？

Answer 1

如果你有一个 IObservable<IDisposable> source然后执行此操作以自动处理先前的值并在序列结束时进行清理:

IObservable<IDisposable> query =
    Observable.Create<IDisposable>(o =>
    {
        var serial = new SerialDisposable();
        return new CompositeDisposable(
            source.Do(x => serial.Disposable = x).Subscribe(o),
            serial);
    })