synchrone Anrufe mit rxjava Android

Question

Meine App hat eine SearchView. Wenn der Benutzer das SearchView eingibt, übergibt die OnQueryTextChange die Abfrage an den Referenten und dann ruft es die API auf. Ich verwende Retrofit und RxJava für die Anrufe. Die Aufrufe geben eine JSON-Datei mit den Wörtern zurück, die enthalten, was der Benutzer bisher eingegeben hat. Das Problem ist, dass, wenn der Benutzer schnell Buchstaben eingibt und das Netzwerk langsam ist, das SearchView die Ergebnisse nicht auf der Grundlage aller eingegebenen Buchstaben zeigt, sondern vielleicht bis zum vorletzten, weil der letzte Aufruf schneller war, um die Ergebnisse zu erhalten verglichen mit dem vorletzten.

Beispiel: der Benutzer Start Typisierung:

"COU" -> machen einen Aufruf an die API (erst nach 3 Buchstaben rufen) -> starten returnin

Werte

"n" -> make a rufen -> Startwert Rückkehr

"t" -> Anruf -> Startwert

"r" Rückkehr -> Anruf (die Verbindung langsam ist)

„y“ -> Anruf -> Startwerte Rückkehr

-> „r“ endlich die Ergebnisse erhalten und die Erträge sie

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

Der Anruf ist sehr einfach und wenn ich erhalte eine Fehlermeldung I will nichts anzeigen.

Gibt es eine Möglichkeit, das zu lösen? Oder ist dies bei reaktiver Programmierung nicht der Fall?

EDIT: nur klarer zu machen, der Fluss ist die folgende:

1. Aktivität, die eine benutzerdefinierte Suche verwendet (https://github.com/Mauker1/MaterialSearchView)

2. der Brauch Suche hat einen Hörer, wenn der Benutzer beginnt zu tippen. Sobald der Benutzer beginnt, die Aktivität einzugeben, ruft der Presenter auf.

3. der Moderator wird eine beobachtbare zurück vom Interaktorsystem abonnieren:

Moderator:

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

Interaktorsystem:

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

So, jetzt entweder ich die benutzerdefinierte Suchansicht ändern in einer 'normalen' Suchansicht und dann RxBinding verwenden oder vielleicht sollte ich einen Handler verwenden o r so etwas (aber immer noch zu kämpfen, wie es in meiner Architektur passen)

Answer 1

Sie sind im Glück gibt es einen Betreiber für das genannte debounce

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

Was debounce ist nicht N Zeiteinheiten für mehr Ergebnisse warten bevor Sie fortfahren. Sagen Sie zum Beispiel, das Netzwerk braucht 2 Sekunden, um zurückzukehren, und Sie fluten es mit Anfrage nach Anfrage, debunce wird 3 Sekunden lang ohne Ergebnisse warten und dann das letzte Ergebnis zurückgeben. Stellen Sie sich vor, dass alles bis auf die N-Zeit der Inaktivität fallen gelassen wird.

dies Ihr Problem lösen, aber immer noch das Netzwerk überflutet, im Idealfall würden Sie die ausgezeichnete RxBinding Bibliothek verwenden Sie die Zurückstellungs vor wie die Anforderung, etwas zu machen:

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

Mit dem aktuellen Setup wird es 3 Sekunden warten nachdem ein Benutzer etwas eingibt und erst dann den Netzwerkanruf tätigt. Wenn sie stattdessen etwas Neues eingeben, wird die erste ausstehende Suchanforderung gelöscht.

Edit: geändert RxTextView.textChanges (Textview), basierend auf OP nicht eine Android-Suche Widget

Answer 2

Erweiterung über die Verwendung, was @MikeN gesagt, wenn Sie nur wollen, um die Ergebnisse der letzten Eingabe verwenden, sollten Sie verwenden switchMap() (das ist flatMapLatest() in einigen anderen Rx-Implementierungen).

Answer 3

Ich habe das Überflutungsproblem gelöst, ohne RxBinding zu verwenden, und ich möchte meine Lösung posten, nur für den Fall, dass jemand anderes es braucht. Also immer wenn der onTextChanged aufgerufen wird, überprüfe ich zunächst, ob die Größe> 2 ist und ob es mit dem Netzwerk verbunden ist (Boolean von einem BroadcastReceiver aktualisiert). Dann habe ich die zu sendende Nachricht verzögert und lösche alle anderen Nachrichten in der Warteschlange. Das bedeutet, dass ich nur die Abfragen ausführen wird, die nicht innerhalb der angegebenen Verzögerung sind:

@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); 
     } 
    } 

Dann wird der 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); 
     } 
    } 
};