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 ... in LINQ comprehension转换为a 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();

我强烈建议您查看Rx Hands on Lab,它可以更详细地解释这一点.