RxJava: установить тайм-аут для нескольких наблюдаемых?

Question

У меня есть сценарий, в котором у меня есть два наблюдаемых, каждый из которых возвращает элемент из вызова API.

То, что я хочу сделать, это:

• Выполнить и апи вызовы одновременно.
• Установите значение тайм-аута (500 миллисекунд) для обоих методов.
• Получите результат как после их завершения, если оба из них заняли меньше значения таймаута (500 миллисекунд), я получу результаты от обоих, если один заканчивается, то я хочу получить его результат и получить ошибку или нулевое значение второго.
• Если оба пользователя не закончили в течение таймаута, я хочу получить сообщение об ошибке.

Я ценю любые подсказки о том, как достичь этого.

Answer 1

Предполагая, что оба вашего api возвращают Observable, который имеет выбросы типа T, первый шаг - это изменение этих наблюдаемых, чтобы они тайм-аут. Вы можете сделать это, используя функции amb и Timer.

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")))); 

Второй шаг - дальнейшее изменение ваших наблюдаемых данных, чтобы они выполняли свою работу на разных потоках. Это делается с помощью subscribeOn оператора

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

Последний шаг заключается в объединении выбросов из ваших наблюдаемых, но вы хотите какие-либо ошибки тайм-аута, издаваемый последним, к счастью mergeDelayError оператор делает только, что так

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