我是想实现一个数据库表即时搜索用10000+记录。

I was trying to implement instant search on a database table with 10000+ records.

搜索开始的时候里面的搜索文本框中的文本更改，当搜索框为空我想调用不同的方法加载的所有数据。

The search starts when the text inside the search text box changes, when the search box becomes empty I want to call a different method that loads all the data.

此外，如果用户改变，而正在加载结果其他搜索搜索字符串时，这些结果则装载应停止支持新的搜索。

Also if the user changes the search string while results for another search are being loaded, then the loading of the those results should stop in favor of the new search.

我实现它像下面code，但我不知道是否有更好或更清洁的方式使用接收（无扩展名），运营商这样做，我觉得创建第二个观察到的内部认购的方法首先观察到的是比声明更为迫切，同样对于if语句。

I implemented it like the following code, but I was wondering if there is a better or cleaner way to do it using Rx (Reactive Extension) operators, I feel that creating a second observable inside the subscribe method of the first observable is more imperative than declarative, and the same for that if statement.

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 方法会返回一个的IEnumerable＆LT;党＆GT; 使用SQLite从数据读取器读取数据

The SearchByNameAndNotes method just returns an IEnumerable<Party> using SQLite by reading data from a data reader.

推荐答案

我想你想是这样的。编辑：从您的意见，我看你有一个同步的存储库的API - 我会离开的异步版本中，事后添加同步版本。注在线：

I think you want something like this. From your comments, I see you have a synchronous repository API - I'll leave the asynchronous version in, and add a synchronous version afterwards. Notes inline:

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

An asynchronous repository interface could be something like this:

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

然后我重构查询为：

Then I refactor the query as:

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 => { },
    () => { });

同步存储库版本

这是同步的资料库界面可能是这样的：

Synchronous Repository Version

An synchronous repository interface could be something like this:

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

我个人不推荐一库接口是同步的这样。为什么？它通常会做的IO，所以你会浪费阻塞线程。

Personally, I don't recommend a repository interface be synchronous like this. Why? It is typically going to do IO, so you will wastefully block a thread.

您可能会说，客户可以从后台线程中调用，或者你可以包装在一个任务，他们的呼叫 - 但这不是正确的方式去我觉得

You might say the client could call from a background thread, or you could wrap their call in a task - but this is not the right way to go I think.

在客户端不知道你要阻止;它不是前pressed合同 这应该是处理的执行异步方面的资源库 - 毕竟，如何的，这是最好的实现只会最好的资源库实施者被称为 The client doesn't "know" you are going to block; it's not expressed in the contract It should be the repository that handles the asynchronous aspect of the implementation - after all, how this is best achieved will only be known best by the repository implementer.

总之，接受实施以上，一种方法是这样的（当然它主要是相似的异步版本，所以我只注明了差异）：

Anyway, accepting the above, one way to implement is like this (of course it's mostly similar to the async version so I've only annotated the differences):

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 => { },
    () => { });