使用响应式扩展搜索 TextChanged

Question

我试图在包含 10000 多条记录的数据库表上实现即时搜索。

当搜索文本框内的文本发生变化时开始搜索，当搜索框变空时我想调用一个不同的方法来加载所有数据。

此外，如果用户在加载另一个搜索的结果时更改了搜索字符串，那么这些结果的加载应该停止以支持新的搜索。

我像下面的代码一样实现了它，但我想知道是否有更好或更清洁的方法来使用 Rx（反应性扩展）运算符来实现它，我觉得在第一个 observable 的 subscribe 方法中创建第二个 observable 是比声明式更具命令性，if 语句也是如此。

var searchStream = Observable.FromEventPattern(s => txtSearch.TextChanged += s, s => txtSearch.TextChanged -= s)
    .Throttle(TimeSpan.FromMilliseconds(300))
    .Select(evt =>
        {
            var txtbox = evt.Sender as TextBox;
            return txtbox.Text;
        }
    );

searchStream
    .DistinctUntilChanged()
    .ObserveOn(SynchronizationContext.Current)
    .Subscribe(searchTerm =>
        {
            this.parties.Clear();
            this.partyBindingSource.ResetBindings(false);
            long partyCount;
            var foundParties = string.IsNullOrEmpty(searchTerm) ? partyRepository.GetAll(out partyCount) : partyRepository.SearchByNameAndNotes(searchTerm);

            foundParties
                .ToObservable(Scheduler.Default)
                .TakeUntil(searchStream)
                .Buffer(500)
                .ObserveOn(SynchronizationContext.Current)
                .Subscribe(searchResults =>
                    {
                        this.parties.AddRange(searchResults);
                        this.partyBindingSource.ResetBindings(false);
                    }
                    , innerEx =>
                    {

                    }
                    , () => { }
                );
        }
        , ex =>
        {
        }
        , () =>
        {

        }
    );

SearchByNameAndNotes 方法只是使用 SQLite 通过从数据读取器读取数据返回一个 IEnumerable<Party>。

Answer 1

我想你想要这样的东西。编辑：从您的 cmets，我看到您有一个同步存储库 API - 我将保留异步版本，然后添加一个同步版本。内联注释：

异步存储库版本

异步存储库接口可能是这样的：

public interface IPartyRepository
{
    Task<IEnumerable<Party>> GetAllAsync(out long partyCount);
    Task<IEnumerable<Party>> SearchByNameAndNotesAsync(string searchTerm);
}

然后我将查询重构为：

var searchStream = Observable.FromEventPattern(
    s => txtSearch.TextChanged += s,
    s => txtSearch.TextChanged -= s)
    .Select(evt => txtSearch.Text) // better to select on the UI thread
    .Throttle(TimeSpan.FromMilliseconds(300))
    .DistinctUntilChanged()
    // placement of this is important to avoid races updating the UI
    .ObserveOn(SynchronizationContext.Current)
    .Do(_ =>
    {
        // I like to use Do to make in-stream side-effects explicit
        this.parties.Clear();
        this.partyBindingSource.ResetBindings(false);
    })
    // This is "the money" part of the answer:
    // Don't subscribe, just project the search term
    // into the query...
    .Select(searchTerm =>
    {
        long partyCount;
        var foundParties = string.IsNullOrEmpty(searchTerm)
            ? partyRepository.GetAllAsync(out partyCount)
            : partyRepository.SearchByNameAndNotesAsync(searchTerm);

        // I assume the intention of the Buffer was to load
        // the data into the UI in batches. If so, you can use Buffer from nuget
        // package Ix-Main like this to get IEnumerable<T> batched up
        // without splitting it up into unit sized pieces first
        return foundParties
            // this ToObs gets us into the monad
            // and returns IObservable<IEnumerable<Party>>
            .ToObservable()
            // the ToObs here gets us into the monad from
            // the IEnum<IList<Party>> returned by Buffer
            // and the SelectMany flattens so the output
            // is IObservable<IList<Party>>
            .SelectMany(x => x.Buffer(500).ToObservable())
            // placement of this is again important to avoid races updating the UI
            // erroneously putting it after the Switch is a very common bug
            .ObserveOn(SynchronizationContext.Current); 
    })
    // At this point we have IObservable<IObservable<IList<Party>>
    // Switch flattens and returns the most recent inner IObservable,
    // cancelling any previous pending set of batched results
    // superceded due to a textbox change
    // i.e. the previous inner IObservable<...> if it was incomplete
    // - it's the equivalent of your TakeUntil, but a bit neater
    .Switch() 
    .Subscribe(searchResults =>
    {
        this.parties.AddRange(searchResults);
        this.partyBindingSource.ResetBindings(false);
    },
    ex => { },
    () => { });

同步存储库版本

同步存储库接口可能是这样的：

public interface IPartyRepository
{
    IEnumerable<Party> GetAll(out long partyCount);
    IEnumerable<Party> SearchByNameAndNotes(string searchTerm);
}

就个人而言，我不建议存储库接口像这样同步。为什么？它通常会做 IO，所以你会浪费地阻塞一个线程。

您可能会说客户端可以从后台线程调用，或者您可以将他们的调用包装在一个任务中 - 但我认为这不是正确的方法。

• 客户端不“知道”你要阻止；合同中没有写明
• 应该由存储库来处理实现的异步方面 - 毕竟，如何最好地实现这一点只有存储库实施者才能最好地了解。

无论如何，接受上述，一种实现方式是这样的（当然它与异步版本大多相似，所以我只注释了差异）：

var searchStream = Observable.FromEventPattern(
    s => txtSearch.TextChanged += s,
    s => txtSearch.TextChanged -= s)
    .Select(evt => txtSearch.Text)
    .Throttle(TimeSpan.FromMilliseconds(300))
    .DistinctUntilChanged()
    .ObserveOn(SynchronizationContext.Current)
    .Do(_ =>
    {
        this.parties.Clear();
        this.partyBindingSource.ResetBindings(false);
    })       
    .Select(searchTerm =>
        // Here we wrap the synchronous repository into an
        // async call. Note it's simply not enough to call
        // ToObservable(Scheduler.Default) on the enumerable
        // because this can actually still block up to the point that the
        // first result is yielded. Doing as we have here,
        // we guarantee the UI stays responsive
        Observable.Start(() =>
        {
            long partyCount;
            var foundParties = string.IsNullOrEmpty(searchTerm)
                ? partyRepository.GetAll(out partyCount)
                : partyRepository.SearchByNameAndNotes(searchTerm);

            return foundParties;
        }) // Note you can supply a scheduler, default is Scheduler.Default
        .SelectMany(x => x.Buffer(500).ToObservable())
        .ObserveOn(SynchronizationContext.Current))
    .Switch()
    .Subscribe(searchResults =>
    {
        this.parties.AddRange(searchResults);
        this.partyBindingSource.ResetBindings(false);
    },
    ex => { },
    () => { });