可觀察的取消令牌

Question

如果在StartButton單擊中，即從停止按鈕創建以下可觀察項，我如何取消以下類型的Rx Observable。

var instance = ThreadPoolScheduler.Instance; 

Observable.Interval(TimeSpan.FromSeconds(2), instance) 
        .Subscribe(_ => 
        { 
        Console.WriteLine(DateTime.Now); // dummy event 
        } 
        );   

Answer 1

只需使用的Subscribe，需要一個重載的一個CancellationToken：

observable.Subscribe(_ => Console.WriteLine(DateTime.UtcNow), cancellationToken); 

這簡化喬恩斯基特的例子：

using System; 
using System.Reactive.Concurrency; 
using System.Reactive.Linq; 
using System.Threading; 

class Program 
{ 
    static void Main(string[] args) 
    { 
     var instance = ThreadPoolScheduler.Instance; 
     var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); 

     Observable.Interval(TimeSpan.FromSeconds(0.5), instance) 
      .Subscribe(_ => Console.WriteLine(DateTime.UtcNow), cts.Token); 
     Thread.Sleep(10000); 
    } 
} 

Answer 2

您保留，是由Subscribe返回IDisposable，並在其上調用Dispose。

很可能有一種方法可以將基於Rx IDisposable的取消訂閱與CancellationToken開箱即用，但只需撥打Dispose即可。 （你總是可以只註冊一個延續與取消標記調用dispose ...）

下面是一個簡短但完整的例子來說明：

using System; 
using System.Reactive.Concurrency; 
using System.Reactive.Linq; 
using System.Threading; 

class Program 
{ 
    static void Main(string[] args) 
    { 
     var instance = ThreadPoolScheduler.Instance; 
     var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); 

     var disposable = Observable 
      .Interval(TimeSpan.FromSeconds(0.5), instance) 
      .Subscribe(_ => Console.WriteLine(DateTime.UtcNow)); 
     cts.Token.Register(() => disposable.Dispose()); 
     Thread.Sleep(10000); 
    } 
}