为什么这个CancellationDisposable永远不会在Observable.Dispose()上被取消

Question

我使用RxFramework在一个WinForms应用程序。我试图运行一个可观察对象异步和使用CancellationDisposable取消操作，当用户点击一个按钮。但这行不通!

假设我有一个表单，有两个按钮和一个进度条。点击订阅观察者在一个新的线程。然后按下Button2_click以取消操作。为什么 cancel.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线程阻塞
在ThreadPool线程上，执行进入lambda
3. CancellationDisposable已创建
4. 您多次检查取消并执行整个过程。在此期间，您的进度条将被更新。
5. cancel从lambda
返回
6. UI线程被释放，obs获得其值
7. 单击Button2，呼叫obs.Dispose
8. 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

不如这样，上面的代码有很多bug，但实际上有一个更好的方法来完成这个(警告:在前面的TextArea中编码):

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 */);