system.reactive - Reactive Extensions Switch() 在搜索中

Question

经典实时搜索示例:

var searchResults = from input in textBoxChanged
                    from results in GetDataAsync(input)
                    select results;

GetDataAsync 返回:

 Task<List<DataRecord>>

这显然是一个竞争条件，因此如果由第二个输入触发的搜索在第一个输入之前返回，则第一个输入的结果在后，因此给我错误的数据。

我到处读到 .Switch() 运算符会神奇地解决这个问题，但是如何解决？

.Switch() 只存在于:

 IObservable<IObservable<T>>

Answer 1

假设

我假设 textBoxChanged 是由以下内容创建的:

var textBoxChanged = Observable.FromEventPattern(x, "TextChanged")
                               .Select(evt => ((TextBox)evt.Sender).Text);

使用 SelectMany 时防止竞争条件

from... from... 在 LINQ 理解中转换为 SelectMany ，这就是您正在使用的。 Rx 足够聪明，可以翻译 Task<List<DataRecord>>返回者 GetDataAsync(input)成 IObservable<List<DataRecord>> .

问题是您想防止结果从最近提交的搜索请求之外的所有请求返回。

为此，您可以利用 TakeUntil .它具有以下签名:

public static IObservable<TSource> TakeUntil<TSource, TOther>(
    this IObservable<TSource> source,
    IObservable<TOther> other
)

并且它从源可观察序列返回值，直到另一个可观察序列产生一个值。

我们可以这样使用它:

var searchResults = from input in textBoxChanged
                    from results in GetDataAsync(input).ToObservable().TakeUntil(textBoxChanged)
                    select results;

这将防止竞争条件，但也会订阅 textBoxChanged 两次。

使用 Switch反而

这是一种非常有用的模式，使用 Switch() 引入了一种替代方法。也负责双重订阅的运营商。

而不是使用 SelectMany ，只需将输入直接投影到搜索查询中 - 这将给出 IObservable<IObservable<List<DataRecord>> 的返回类型，一个流。 Switch 会从一个流跳转到另一个流，只返回最近的流。这相当于 SelectMany/TakeUntil 组合:

var searchResults = (from input in textBoxChanged
                     select GetSearchResults(input).ToObservable())
                    .Switch();

I highly suggest looking at the Rx Hands on Lab that explains this in much more detail.