如何在Android中压缩两个Observable？

Question

在我的应用程序中，我有两个服务，它们都有一个发出请求的方法，然后返回一个不同类型的Observable。

我想在RecyclerView中显示一个由组合这两个Observable的结果组成的列表。我搜索了这个，发现zip（）方法似乎完全符合我的要求。我正在尝试实现它，但我不知道如何正确地执行它。

当我在谷歌搜索时，我想出了这个this article，似乎清楚地解释了它。即使作者在使用Observables时使用的是Singles。

据我所知，zip()是如何工作的，我知道我必须传递每个我想要“压缩”的Observable，然后我必须指定一个函数来组成我的最终Observable，对吧？

到目前为止这是我的代码：

interface InvitationService {
    @GET("foo/{userId}")
    fun getFooByUser(@Path("userId") userId: String): Observable<Response<ArrayList<Foo>>>
}

interface InvitationService {
    @GET("bar/{userId}")
    fun getBarByUser(@Path("userId") userId: String): Observable<Response<ArrayList<Bar>>>
}

class FooRemoteDataSource : FooDataSource {
    private var apiService: FooService

    fun getFooByUser(userId:String) {
        return apiService.getFooByUser(userId)
    }
}

class BarRemoteDataSource : BarDataSource {
    private var apiService: BarService

    fun getBarByUser(userId:String) {
        return apiService.getBarByUser(userId)
    }

}

class FooRepository(private val remoteDataSource: InvitationRemoteDataSource) : FooDataSource {

    override fun getFooByUser(userId: String): Observable<Response<ArrayList<Foo>>> {
        return remoteDataSource.getFooByUser(userId)
    }
}

class BarRepository(private val remoteDataSource: BarRemoteDataSource) : BarDataSource {

    override fun getBarByUser(userId: String): Observable<Response<ArrayList<Bar>>> {
        return remoteDataSource.getBarByUser(userId)
    }
}

这就是我实际上被困住的地方：

class ListPresenter(var listFragment: ListContract.View?, 
                        val fooRepository: FooRepository,
                        val barRepository: BarRepository) : ListContract.Presenter {

    fun start() {
        loadList()
    }

    private fun loadLists() {
        //HERE IS WHERE IM STUCK

        Observable.zip(fooRepository.getFooByUser(userId).subscribeOn(Schedulers.io()),
                barRepository.getBarByUser(userId).subscribeOn(Schedulers.io()),
                )

        // AFTER 'ZIPPING' THE OBSERVABLES 
        // I NEED TO UPDATE THE VIEW ACCORDINGLY
    }
}

我不知道如何恰当地调用zip()，我知道我必须通过一个函数，但是我没有得到它，因为在上面链接的文章中，作者使用的是Function3，因为他有3个Observables。

因为我只有2，我不知道该怎么做。如果在方法args中的逗号之后打开花括号，它需要我返回一个BiFunction<ArrayList<Foo>, ArrayList<Bar>>，这是我不知道如何指定。

有人会向我解释一下吗？

Answer 1

对于Kotlin，你应该使用RxKotlin而不是RxJava。 BiFunction，Function3来自RxJava。使用RxKotlin，您可以使用lambdas。

---

据我所知，zip（）的工作原理，我知道我必须传递每个Observable我想要“zip”然后我必须指定一个函数来组成我的最终Observable，对吧？

正确，这是一个最小的例子，演示了如何做到这一点。

例1

val observable1 = listOf(1, 2, 3).toObservable()
val observable2 = listOf(4, 5, 6).toObservable()
val zipped = Observables.zip(observable1, observable2) { o1, o2 -> o1 * o2}

在这个例子中，你有两个observable，每个都发出整数。你将它们传递给zip，并作为第三个参数传递一个lambda，它定义了一种“cobmine”的方法。在这种情况下，它会倍增它们。

由此产生的可观察的zipped将发出：4,10和18。

---

例2

这里的另一个例子是压缩三个不同类型的observable：

val obs1 = listOf("on", "tw", "thre").toObservable()
val obs2 = listOf("n", "o", "e").toObservable()
val obs3 = listOf(1, 2, 3).toObservable()
val zipped = Observables.zip(obs1, obs2, obs3) { o1, o2, o3 -> 
    "$o1$o2 = $o3"
}

这里，得到的observable的每个元素都是一个字符串：“one = 1”，“two = 2”，“three = 3”

Answer 2

使用BiFunction压缩两个不同类型的Observable

override fun getCommoditiesAndAddresses() {
    view.showProgress()
    view.hideViews()
    Observable.zip(Commo24Retrofit.createAuthService(RateAPIService::class.java)
            .getCommodities(),
            Commo24Retrofit.createAuthService(RateAPIService::class.java)
                    .getLocations(GetLocationsRequest(getOrgId())),
            BiFunction { commodityResponse: GetCommoditiesResponse, locationsResponse: GetLocationsResponse ->
                handleCommoditiesAndAddresses(commodityResponse, locationsResponse)
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                view.hideProgress()
                view.showViews()
                view.handleCommodities(it?.commodities)
                view.handleLocations(it?.locations)
            }, { throwable ->
                view.hideProgress()
                view.handleFailure(throwable.getErrorMessage(context))
            })
}

看，我是如何处理响应的：

private fun handleCommoditiesAndAddresses(commodityResponse: GetCommoditiesResponse, locationsResponse: GetLocationsResponse): CommoditiesAddresses {
        return CommoditiesAddresses(commodityResponse.commodityList, locationsResponse.addressList)
    }

在这里，查看API服务：

interface RateAPIService {

    @POST("get-org-address")
    fun getLocations(@Body getLocationsRequest: GetLocationsRequest): Observable<GetLocationsResponse>

    @POST("get-commodity-list")
    fun getCommodities(): Observable<GetCommoditiesResponse>
    }

如果您有任何疑问，可以将其评论出来。