синхронные звонки с rxjava Android

Question

У моего приложения есть SearchView. Когда пользователь вводит в SearchView, onQueryTextChange передает запрос ведущему, а затем вызывает API. Я использую Retrofit и RxJava для вызовов. Вызовы возвращают json-файл со словами, содержащими то, что пользователь набрал до сих пор. Проблема заключается в том, что если пользователь быстро набирает буквы, а сеть работает медленно, иногда SearchView не отображает результаты на основе всех введенных букв, но, возможно, до второго, потому что последний вызов был быстрее, чтобы получить результаты по сравнению со вторым последним.

Пример: начало пользователь набрав:

"Коу" -> сделать вызов к API (сначала вызвать после 3-х букв) -> начать возвращаться д значения

"п" -> сделать звоните -> начать возвращать значения

"т" -> сделать вызов -> начать возвращать значения

"г" -> сделать вызов (соединение медленное)

«у» -> сделать вызов -> начать возвращать значения

-> «г» получить результаты, наконец, и возвращает их

public Observable<List<MyModel>> getValues(String query) { 
    return Observable.defer(() -> mNetworkService.getAPI() 
      .getValues(query) 
      .retry(2) 
      .onErrorReturn(e -> new ArrayList<>())); 
} 

вызов очень прост и всякий раз, когда я получаю I ошибки не хотят ничего показывать.

Есть ли способ решить это? Или, возможно, это не так, чтобы использовать реактивное программирование?

EDIT: Просто, чтобы сделать более ясным, поток выглядит следующим образом:

1. Деятельность, которая использует вид пользовательского поиска (https://github.com/Mauker1/MaterialSearchView)

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

3. ведущий подпишется наблюдаемым возвращаемый интерактора:

Ведущий:

addSubscription(mInteractor.getValues(query) 
      .observeOn(mMainScheduler) 
      .subscribeOn(mIoScheduler) 
      .subscribe(data -> { 
       getMvpView().showValues(data); 
      }, e -> { 
       Log.e(TAG, e.getMessage()); 
      })); 

Interactor:

public Observable<List<MyModel>> getValues(String query) { 
    return Observable.defer(() -> mNetworkService.getAPI() 
      .getValues(query) 
      .debounce(2, TimeUnit.SECONDS) 
      .retry(2) 
      .onErrorReturn(e -> new ArrayList<>())); 

Так что теперь я или изменить вид пользовательского поиска в «нормальном» поиске, а затем использовать RxBinding или, может быть, я должен использовать обработчик o г что-то подобное (но по-прежнему изо всех сил, как вписать его в моей архитектуры)

Answer 1

Вы находитесь в удаче есть оператор для этого называется дребезга

Observable.defer(() -> mNetworkService.getAPI() 
      .getValues(query) 
      .debounce(3, TimeUnit.SECONDS) 
      .retry(2) 
      .onErrorReturn(e -> new ArrayList<>())); 

Что дребезга делает это ждать единицы времени N для получения дополнительных результатов до продолжения. Скажем, например, что сеть занимает 2 секунды, чтобы вернуться, и вы заливаете ее запросом после запроса, debounce будет ждать 3 секунды без результатов и затем возвращает последний результат. Подумайте об этом как о том, чтобы сбросить все, кроме одного до N времени бездействия.

Это решить вашу проблему, но все-таки будет переполнять сеть, в идеале вы бы использовать прекрасную библиотеку RxBinding сделать Отложить до принятия запроса что-то вроде:

RxTextView.textChanges(searchView) 
.debounce(3, TimeUnit.SECONDS) 
.map(input->mNetworkService.getAPI().getValues(input.queryText().toString())) 
.retry(2) 
.onErrorReturn(e -> new ArrayList<>())) 

С текущей настройки он будет ждать 3 секунды после того, как пользователь что-то наберет и только затем выполнит сетевой вызов. Если вместо этого они начинают вводить что-то новое, первый ожидающий запрос поиска отбрасывается.

Edit: изменен RxTextView.textChanges (TextView) на основе ОП, не использующих андроид SearchView виджет

Answer 2

Расширение на то, что сказал @MikeN, если вы хотите использовать только результаты последнего ввода, вы должны использовать switchMap() (то есть flatMapLatest() в некоторых других реализациях Rx).

Answer 3

Я решил проблему затопления без использования RxBinding, и я хочу опубликовать свое решение на случай, если кому-то это понадобится. Итак, всякий раз, когда вызывается onTextChanged, я проверяю, в первую очередь, если размер> 2 и если он подключен к сети (булев, обновляемый BroadcastReceiver). Затем я создаю сообщение для отправки, которое задерживается, и я удаляю все остальные сообщения в очереди. Это означает, что я буду выполнять только те запросы, которые не являются в определенной задержкой:

@Override 
    public void onTextChanged(CharSequence s, int start, int before, int count) { 

     if (TextUtils.getTrimmedLength(s) > 2 && isConnected) { 
      mHandler.removeMessages(QUERY_MESSAGE); 
      Message message = Message.obtain(mHandler, QUERY_MESSAGE, s.toString().trim()); 
      mHandler.sendMessageDelayed(message, MESSAGE_DELAY_MILLIS); 
     } 
    } 

Тогда Handler:

private Handler mHandler = new Handler() { 
    @Override 
    public void handleMessage(Message msg) { 
     if (msg.what == QUERY_MESSAGE) { 
      String query = (String)msg.obj; 
      mPresenter.getValues(query); 
     } 
    } 
};