为什么这个CancellationDisposable永远不会取消Observable.Dispose（）？

Question

在WinForms应用程序中使用RxFramework。我试图运行Observable异步并使用CancellationDisposable来取消操作时，用户单击按钮。但它不工作！

假设我有一个带有2个按钮和ProgressBar的窗体。 Button1_click订阅新线程的观察者。然后立即按下Button2_click以取消操作。 为什么取消.Token.IsCancellationRequested永远不会是真的？

private IDisposable obs = null; 
private void button1_Click(object sender, EventArgs e) { 
    var countObserver = Observable.Create<int>(observer => { 
     var cancel = new CancellationDisposable(); 

     if (!cancel.Token.IsCancellationRequested) { 
      //Step 1 of a long running process using lot of resources... 
      observer.OnNext(1); 
     } 
     if (!cancel.Token.IsCancellationRequested) { 
      //Step 2 of a long running process using lot of resources... 
      observer.OnNext(1); 
     } 
     if (!cancel.Token.IsCancellationRequested) { 
      //Step 3 of a long running process using lot of resources... 
      observer.OnNext(1); 
     } 
     observer.OnCompleted(); 

     return cancel; 
    }); 

    obs = countObserver 
     .ObserveOn(new ControlScheduler(this)) 
     .SubscribeOn(Scheduler.ThreadPool) 
     .Subscribe(i => { 
      //Update a progress bar here... 
     }); 

} 

private void button2_Click(object sender, EventArgs e) { 
    if (obs != null) 
     obs.Dispose(); 
} 

Answer 1

它发生这种情况是因为您传递给Observable.Create的lambda不会返回CancellationDisposable直到它会经历所有步骤。因此，动作的顺序如下：

1. 你叫Subscribe
2. UI线程块
3. 在一个ThreadPool线程，执行进入拉姆达创建
4. 您检查
5. CancellationDisposable取消几次并执行整个过程。在此期间，您的进度栏会更新。
6. cancel从拉姆达返回
7. UI线程被释放，并obs获取其值
8. 您点击Button2，并呼吁obs.Dispose
9. cancel得到cancel.Token.IsCancellationRequested=true。但是，一切都已经发生了，没有任何反应。

这里的固定码：

private void button1_Click(object sender, EventArgs e) { 
    var countObserver = Observable.Create<int>(observer => { 
    var cancel = new CancellationDisposable(); 

    // Here's the magic: schedule the job in background and return quickly 
    var scheduledItem = Scheduler.ThreadPool.Schedule(() => 
    { 
     if (!cancel.Token.IsCancellationRequested) { 
     //Step 1 of a long running process using lot of resources... 
     observer.OnNext(1); 
     } 
     if (!cancel.Token.IsCancellationRequested) { 
     //Step 2 of a long running process using lot of resources... 
     observer.OnNext(1); 
     } 
     if (!cancel.Token.IsCancellationRequested) { 
     //Step 3 of a long running process using lot of resources... 
     observer.OnNext(1); 
     } 
     observer.OnCompleted(); 
    }); 

    return new CompositeDisposable(cancel, scheduledItem); 
}); 

obs = countObserver 
    .ObserveOn(new ControlScheduler(this)) 
    .Subscribe(i => { 
     //Update a progress bar here... 
    }); 

} 

Answer 2

这个怎么样，而不是，有一些与上面的代码错误的，但实际上，有一种更好的方式来完全做到这一点（警告：在文本区域编码前）：

countObservable = Observable.Timer(new ControlScheduler(this)); 

var buttonObservable = Observable.FromEventPattern<EventArgs>(
    x => button1.Click += x, x => button1.Click -= x); 

countObservable 
    .TakeUntil(buttonObservable) 
    .Subscribe(x => /* Do stuff */);