RxJava: Как я могу сбросить длинную горячую наблюдаемую цепочку?

Question

Для функции поиска моего приложения у меня есть горячая наблюдаемая цепочка, которая делает следующее.

1. Принимает строку ввода пользователя в EditText (а TextChangedEvent) (на mainThread)
2. DEBOUNCE 300мса (на computation нити)
3. Показать загрузки роллерных (mainThread)
4. Опросить SQL дб с этой строкой (этот запрос может занимать от 100 мс до 2000 мс) (на Schedulers.io())
5. Показать результаты для пользователя (mainThread)

Поскольку шаг 3 является настолько переменным по длине, возникает условие гонки, когда отображаются менее свежие результаты поиска по более поздним результатам (иногда). Давайте предположим, что пользователь хочет ввести chicken, но из-за странных скорости набора текста, то первая часть слова испускается до всего срока:

• поиск chick отправляется первым, а затем chicken.
• сказать, что chick принимает 1500ms для выполнения в то время как chicken принимает 300ms для выполнения.
• это приводит к неправильному отображению результатов поиска chick для поискового запроса chicken. Это связано с тем, что сначала был выполнен поиск 10 (занял всего 300 мс), а затем поиск chick (1500 мс).

Как я могу справиться с этим сценарием?

• Как только пользователь запускает новый поиск через TextChangedEvent, мне не нужен старый поиск, даже если он все еще работает. Есть ли способ отменить старый поиск?

Полный наблюдаемым код:

subscription = WidgetObservable.text(searchText) 
       .debounce(300, TimeUnit.MILLISECONDS) 
       .observeOn(AndroidSchedulers.mainThread()) 
         //do this on main thread because it's a UI element (cannot access a View from a background thread) 

         //get a String representing the new text entered in the EditText 
       .map(new Func1<OnTextChangeEvent, String>() { 
        @Override 
        public String call(OnTextChangeEvent onTextChangeEvent) { 
         return onTextChangeEvent.text().toString().trim(); 
        } 
       }) 
       .subscribeOn(AndroidSchedulers.mainThread()) 
       .doOnNext(new Action1<String>() { 
        @Override 
        public void call(String s) { 
         presenter.handleInput(s); 
        } 
       }) 
       .subscribeOn(AndroidSchedulers.mainThread()) 
       .observeOn(Schedulers.io()) 
       .filter(new Func1<String, Boolean>() { 
        @Override 
        public Boolean call(String s) { 
         return s != null && s.length() >= 1 && !s.equals(""); 
        } 
       }).doOnNext(new Action1<String>() { 
        @Override 
        public void call(String s) { 
         Timber.d("searching for string: '%s'", s); 
        } 
       }) 
         //run SQL query and get a cursor for all the possible search results with the entered search term 
       .flatMap(new Func1<String, Observable<SearchBookmarkableAdapterViewModel>>() { 
        @Override 
        public Observable<SearchBookmarkableAdapterViewModel> call(String s) { 
         return presenter.getAdapterViewModelRx(s); 
        } 
       }) 
       .subscribeOn(Schedulers.io()) 
         //have the subscriber (the adapter) run on the main thread 
       .observeOn(AndroidSchedulers.mainThread()) 
         //subscribe the adapter, which receives a stream containing a list of my search result objects and populates the view with them 
       .subscribe(new Subscriber<SearchBookmarkableAdapterViewModel>() { 
        @Override 
        public void onCompleted() { 
         Timber.v("Completed loading results"); 
        } 

        @Override 
        public void onError(Throwable e) { 
         Timber.e(e, "Error loading results"); 
         presenter.onNoResults(); 
         //resubscribe so the observable keeps working. 
         subscribeSearchText(); 
        } 

        @Override 
        public void onNext(SearchBookmarkableAdapterViewModel searchBookmarkableAdapterViewModel) { 
         Timber.v("Loading data with size: %d into adapter", searchBookmarkableAdapterViewModel.getSize()); 
         adapter.loadDataIntoAdapter(searchBookmarkableAdapterViewModel); 
         final int resultCount = searchBookmarkableAdapterViewModel.getSize(); 
         if (resultCount == 0) 
          presenter.onNoResults(); 
         else 
          presenter.onResults(); 
        } 
       }); 

Answer 1

Использование switchMap вместо flatMap. Это заставит его выкинуть * предыдущий запрос всякий раз, когда вы начинаете новый запрос.

* Как это работает:

Всякий раз, когда внешний источник наблюдаемым производит новое значение, switchMap называет свой селектор, чтобы вернуть новый внутренний наблюдаемую (presenter.getAdapterViewModelRx(s) в данном случае). switchMap затем не подписывается от предыдущего внутреннего наблюдаемого, которого он слушал, и подписывает на новый.

отподписывания от предыдущего внутреннего наблюдаемого имеет два эффекта:

1. Любое уведомление (значение, завершение, ошибка и т.д.), производимая наблюдаемый будет игнорироваться и выбрасывались.

2. Наблюдаемый будет уведомлен о том, что его наблюдатель отказался от подписки и может необязательно принять меры для отмены любого асинхронного процесса, который он представляет.

ли ваши заброшенные запросы фактически отменены или не полностью зависит от реализации presenter.getAdapterViewModelRx(). В идеале они будут отменены, чтобы избежать ненужного траты ресурсов сервера. Но даже если они продолжают работать, № 1 выше, предотвращает появление устаревшего результата вашего кода типа.