如何将 dispose 操作设置为 Observable？

Question

我知道如何创建一个 observable 并分配一个处置动作：

Observable.Create(o =>
{
    // o.OnNext etc.
    return Disposable.Create(() => { /* ... */ });
});

但现在我从查询语法中生成了一个 observable：

var observable = from x in otherObservable
                 select x;

如何为此类查询分配处置操作？

Answer 1

如果我理解正确，您希望在处理订阅时“链接”或“收听”。一种方法是使用IObservable<T> 的Finally 运算符，如下所示：

var ob = from x in Observable.Interval(TimeSpan.FromSeconds(1))
            select x;

// Use Finally to create an intermediate IObservable
var disposeSub = ob.Finally(() => Console.WriteLine("disposed"));

// Subscribe to the intermediate observable instead the original one
var yourSub    = disposeSub.Subscribe(Console.WriteLine);

// Wait for some numbers to print
Thread.Sleep(TimeSpan.FromSeconds(4));

// "disposed" will be written on the console at this point
yourSub.Dispose();

希望有帮助！

Answer 2

我认为你应该澄清你的问题。您所说的“处置行动”是什么意思并不完全清楚。

已建议使用Observable.Finally 调用操作，但此操作将在满足以下第一个条件时运行：

• Observable 发送OnCompleted()
• Observable 发送OnError()
• 订阅句柄已释放。

即您无法保证在订阅句柄上调用Dispose 时该操作会准确执行；它可能已经运行 - 但调用 Dispose 确保它会在调用 Dispose 返回之前被调用。

这可能是您所需要的 - 但相信您的话，您只希望操作在最后一种情况下运行 - 在处理手柄时，您需要附加订阅句柄本身的操作，即：

var otherDisposable = /* your observable */;

Action disposingAction = () => Console.WriteLine("I am disposed!");

var subscription = otherDisposable.Subscribe(/* set your handlers here */);

var disposable = new CompositeDisposable(
    subscription,
    Disposable.Create(disposingAction));

/* The disposingAction is *only* run when this is called */
disposable.Dispose();

我想不出什么场景需要这个，我想知道 Carlos 建议的 Observable.Finally 是否更合适！

Answer 3

你不会处理 observable。你处置了一个 observable 的订阅。

例子：

var observable = from x in otherObservable
             select x;

var sub = observable.Subscribe(DoStuff);
sub.Dispose();