RxJava Flowable.zip从不返回值

Question

我的存储库中有一个代码，它必须调用两个端点。我已经使用了Flowable.zip()，但它似乎没有返回值。即使没有可用的网络，调用也不会失败。

fun fetchRateRemote(): Flowable<ResultWrapper<List<RateModel>>> {
return Flowable.zip<Flowable<CurrenciesDTO>, Flowable<RateDTO>, ResultWrapper<List<RateModel>>>(
{
apiEndpoints.fetchCurrencies(key)
}, {
apiEndpoints.fetchRate(key)
}, { t1, t2 ->
val rateList = mutableListOf<RateModel>()
t2.subscribe { rate->
for((k,v) in rate.quotes ){
val currency = k.removeRange(0,3)
t1.subscribe {cur->
val currencyName = cur.currencies[currency]
if (currencyName != null) {
rateList.add(RateModel("$currencyName ($currency)", v.toString()))
}
}
}
}
ResultWrapper.Success(rateList)
}).subscribeOn(Schedulers.io())
}

我使用包装器来模拟状态，这就是我在视图模型中所做的。

private fun fetchRates(){
disposable.add(repository.fetchRateRemote()
.startWith(ResultWrapper.Loading)
.onErrorReturn {
ResultWrapper.Error(it)
}
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(object : DisposableSubscriber<ResultWrapper<List<RateModel>>>() {
override fun onComplete() {}
override fun onNext(rate: ResultWrapper<List<RateModel>>) {
rates.postValue(rate)
}
override fun onError(error: Throwable) {
error.printStackTrace()
}
})
)
}

然后我通过LiveData观察我的活动率。包装器或观察结果不是问题所在。它与其他电话一起工作，我不知道为什么邮政电话不起作用。我对RxJava相当陌生，所以如果我没有在我的存储库中正确实现某些东西，请帮助纠正我。

Answer 1

好的!我在上面的代码库中犯了很多错误，但我设法修复了它。这是解决方案。zip方法的Type参数错误!我也没有正确地调用bfunction参数

fun fetchRateRemote(): Flowable<ResultWrapper<List<RateModel>>> {
return Flowable.zip<CurrenciesDTO, RateDTO, ResultWrapper<List<RateModel>>>(
apiEndpoints.fetchCurrencies(key), apiEndpoints.fetchRate(key), BiFunction { t1, t2 ->
val rateList = mutableListOf<RateModel>()
for((k,v) in t2.quotes ){
val currencyCode = k.removeRange(0,3)
val currencyName = t1.currencies[currencyCode]
if (currencyName != null) {
rateList.add(RateModel("$currencyName ($currencyCode)", v.toString()))
}
}
ResultWrapper.Success(rateList)
}).subscribeOn(Schedulers.io())
}