Rx和任务

本文介绍了Rx和任务 - 取消正在运行的任务时，新任务催生？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我有我想和接收处理的用户交互场景。

I have an user interaction scenario I'd like to handle with Rx.

该方案类似于经典的当用户停止打字，做一些工作（通常，搜索什么用户目前为止所输入）（1） - 但我还需要：

The scenario is similar to the canonical "when user stops typing, do some work" (usually, search for what the user has typed so far) (1) - but I also need to :

• （2）只能得到做一些工作的结果最新的单位（见下文）
• （3）一个新的工作单位开始时，取消正在进行的任何工作（在我的情况下，它是CPU密集型）

• (2) only get the latest of the results of "do some work" units (see below)
• (3) when a new unit of work starts, cancel any work in progress (in my case it's CPU intensive)

有关（1）我使用的IObservable 为用户的事件，节流与 .Throttle（）仅在触发事件之间的停顿（用户停止输入）。

For (1) I use an IObservable for the user events, throttled with .Throttle() to only trigger on pauses between events ("user stops typing").

这一点，我。选择（_ =方式> CreateMyTask（...）ToObservable（）） 。

这给了我一个的IObservable<的IObservable< T>> ，其中每个内部观测的包装单个任务。

This gives me an IObservable<IObservable<T>> where each of the inner observables wraps a single task.

要获得（2）我终于适用 .Switch（）只得到的最新单位的结果。工作

To get (2) I finally apply .Switch() to only get the results from the newest unit of work.

什么（3） - 取消未完成的任务吗？

What about (3) - cancel pending tasks ?

如果我理解正确的话，每当有一个新的内的IObservable< T> 的 .Switch（）方法订阅它，的取消订阅的与以前的（S），导致它们的Dispose（）。结果也许可以以某种方式连接到触发任务取消？

If I understand correctly, whenever there's a new inner IObservable<T>, the .Switch() method subscribes to it and unsubscribes from the previous one(s), causing them to Dispose(). Maybe that can be somehow wired to trigger the task to cancel?

推荐答案

您可以只使用 Observable.FromAsync 这将产生被取消标记时观察者unsubcribes：

You can just use Observable.FromAsync which will generate tokens that are cancelled when the observer unsubcribes:

input.Throttle(...) .Select(_ => Observable.FromAsync(token => CreateMyTask(..., token))) .Switch() .Subscribe(...);

这将产生每个工作单位新的令牌，每一次取消切换切换到新的。

This will generate a new token for each unit of work and cancel it every time Switch switches to the new one.