Rxjava2 操作符 compose

compose字面意思为组成。Rxjava的链式调用虽然很方便，但在很多时候也显得冗余。

1、比如线程的切换。绝大多时候都是固定的

.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())

2、比如特定格式的数据处理。例如网络请求的返回的数据，通常网络请求返回的数据，我们和后台之间存在某个统一的基类。举个例子：当code = 0的时候，表示后台正确返回了数据，code = 其它表示发生了某类异常。这个时候我们就需要一个统一的判断，用于判断这次的网络请求是否正确返回了我们想要的数据。这个时候compose的作用就能够体现出来了。

//后端返回的数据，主要是 code,msg,data
open class BaseResponse{
    @SerializedName("code")val code: Int? = null@SerializedName("msg")val msg: String? = null
}CommonResponse<out T>:BaseResponse(){
    @SerializedName("data")val data: T? = null
}

以一次完整的网络请求为例：

某个网络请求返回了一个Single<CommonResponse<Result>>.compose(CommonResponseTransformer())  .compose(SchedulersTransformer())其中CommonResponseTransformer 负责数据的基本处理，SchedulersTransformer 负责线程的切换class CommonResponseTransformer <T> : SingleTransformer<CommonResponse<T>, T> {
    override fun apply(upstream: Single<CommonResponse<T>>): SingleSource<T> {
    return upstream.flatMap {
    if (it.code!=0) {
    //抛出一个异常，这样可以在rxjava的异常回调里面处理对应的错误码return@flatMap Single.error<T>(RequestFailedException(it.code))}Single.just(data)}}
}class SchedulersTransformer<T> : SingleTransformer<T, T> {
    override fun apply(upstream: Single<T>): SingleSource<T> {
    return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())}
}