0

我正在 Kotlin 中开发一个 android 应用程序。我有 3 个实时数据 observables。数据来自 Firestore。它们被包裹在 RxJava 的 Observable.combineLatest() 方法中。我想在第一次数据检索时设置超时。

我试图在每个可观察对象上设置超时函数，但是在加载了所有 3 个可观察对象的初始数据后，它们会抛出 TimeoutException。

private fun retrieveAllData() {
Observable.combineLatest(
                retrieveDataObservable1().timeout(10, TimeUnit.SECONDS),
                retrieveDataObservable2().timeout(10, TimeUnit.SECONDS),
                retrieveDataObservable3().timeout(10, TimeUnit.SECONDS),
                Function3<String, String, Boolean, Triple<String, String, Boolean>>
                { firstResult, secondResult, ThirdResult ->
                    Triple(firstResult, secondResult, ThirdResult)
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        { Log.d(TAG, "success") },
                        { throwable -> Log.d(TAG, "error", throwable) }
                )
}

private fun retrieveDataObservable1(): Observable<String> {
        return Observable.create<String> { emitter ->
            val listener = dataRef1.addSnapshotListener { snapshot, e ->
                if (e != null) {
                    emitter.onError(e)
                    return@addSnapshotListener
                }
                emitter.onNext("some value")
            }
         emitter.setCancellable { listener.remove() }
     }
}

如果至少有一个 Observables 在 10 秒内没有发出任何项目，我预计 Observable.combineLatest() 在初始检索时会出现 TimeoutExceiption。如果所有 observables 至少成功发出一次数据，那么就不应该有 TimeoutException。

4

1 回答 1

1

它认为这应该适合你：

import io.reactivex.Observable
import io.reactivex.functions.Function
import io.reactivex.functions.Function3
import java.util.concurrent.TimeUnit

Observable.combineLatest(
    firstObservable,
    secondObservable,
    thirdObservable,
    Function3<String, String, String, String> { first, second, third -> "$first$second$third" })
    .timeout<String, String>(
       Observable.empty<String>().delay(10, TimeUnit.SECONDS),
       Function { Observable.never<String>() }
    )
    .subscribe {
        println("$it")
    }

timeout()运算符的第一个参数是Observable第一项超时。在我们的例子中是 10 秒。对于即将到来的项目，它是一个返回Observable.never()所有项目的函数，因此它永远不会触发超时。

于 2019-09-10T13:35:06.483 回答