反应式扩展开关()在搜索中

Question

经典实时搜索示例：

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

GetDataAsync 返回：

 Task<List<DataRecord>>

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

我一直在到处阅读.Switch（） 运算符会神奇地解决这个问题，但是如何呢？

.开关（） 只存在于：

 IObservable<IObservable<T>>

Answer 1

假设

我假设textBoxChanged是由这样的东西创建的：

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

防止使用"选择多个"时的争用条件

从。。。从。。。在 LINQ 理解中转换为 SelectMany ，这就是您正在使用的。Rx足够聪明，可以将GetDataAsync(input)返回的Task<List<DataRecord>>转换为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>>，一个流的流。切换将从一个流跳转到另一个流，仅返回最近的流。这相当于 SelectMany/TakeUntil 组合：

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

我强烈建议查看 Rx 动手实验室，它对此进行了更详细的解释。