RxJava: Legen Sie eine Zeitüberschreitung für mehrere Observablen fest?

Question

Ich habe ein Szenario, in dem ich zwei Observables habe, jedes gibt ein Element von einem API-Aufruf zurück.

Was ich tun möchte, ist:

• Führen beiden api gleichzeitig aufrufen.
• Legen Sie einen Zeitüberschreitungswert (500 Millisekunden) für beide Methoden fest.
• Empfangen Sie das Ergebnis von beiden, wenn sie fertig sind, wenn beide weniger als den Timeout-Wert (500 Millisekunden) Ich erhalte die Ergebnisse von beiden, wenn man nur beendet dann möchte ich sein Ergebnis erhalten und einen Fehler oder ein erhalten Nullwert des zweiten.
• Wenn beide nicht innerhalb des Timeouts fertig sind, möchte ich einen Fehler erhalten.

Ich schätze alle Hinweise, wie dies zu erreichen ist.

Answer 1

Unter der Annahme, dass beide Ihre API ein Observable zurückgeben, das Emissionen vom Typ T hat, ist der erste Schritt, diese Observablen so zu modifizieren, dass sie auslaufen. Sie können dies mit der amb und Timer Funktion tun.

Observable<T> apiCall1WithTimeout = Observable.amb(apiCall1(), 
                Observable.Timer(500, TimeUnit.MILLISECONDS) 
                  .flatmap(l -> Observable.error(new Exception("Api1 timed out"))); 
Observable<T> apiCall2WithTimeout = Observable.amb(apiCall2(), 
                Observable.Timer(500, TimeUnit.MILLISECONDS) 
                  .flatmap(l -> Observable.error(new Exception("Api2 timed out")))); 

Der zweite Schritt besteht darin, Ihre Observablen so zu modifizieren, dass sie ihre Arbeit an verschiedenen Threads ausführen. Dies geschieht mit dem subscribeOn Operator

Observable<T> apiCall1WithTimeoutNewThread = apiCall1WithTimeout.subscribeOn(Schedulers.newThread()); 
Observable<T> apiCall2WithTimeoutNewThread = apiCall2WithTimeout.subscribeOn(Schedulers.newThread()); 

Der letzte Schritt ist es, die Emissionen von Ihrem Observablen zu verschmelzen, aber Sie wollen keine Timeout-Fehler zuletzt emittiert werden, zum Glück der mergeDelayError Betreiber genau das tun so

Observable<T> mergedCalls = Observable.mergeDelayError(apiCall1WithTimeoutNewThread, apiCall2WithTimeoutNewThread);