如何设置Observable的处理动作？

Question

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

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

但是现在我从查询语法中产生了一个observable:

var observable = from x in otherObservable
                 select x;

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

Answer 1

如果我理解正确,那么无论何时处理订阅,您都希望"链接"或"收听".一种方法是使用这样的Finally运算符IObservable<T>:

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();

希望有所帮助!