java - 使用 compose() 时使用 RxJav/RxKotlin2 自定义 ObservableTransformer

Question

我尝试编写一个用于compose()减少样板代码的转换函数。这很简单：

    fun <R> withSchedulers(): ObservableTransformer<R, R> {
        return ObservableTransformer {
            it.subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread())
        }
    }

因此，每次我想订阅任何内容ioThread并收听结果时mainThread，只需几行代码：

    Observable.just(1)
        .compose(MyUtilClass.withSchedulers())
        .subscribe()

但不仅有，Observable而且我们还有Single,和。所以每次我想把它们和我的函数结合起来时，我都必须把它转换成新的类型（这是我没想到的）。CompletableMaybeFlowablewithSchedulers()

例如，

Completable.fromAction { 
        Log.d("nhp", "hello world")
    }//.compose(MyUtilClass.withSchedulers()) <-- This is not compiled
            .toObservable() <--- I have to transform it into Observable
            .compose(MyUtilClass.withSchedulers())
            .subscribe()

所以我的问题是，有没有办法编写上述函数以compose()用于任何类型的Observable( Single, Completable,...) ？或者我们必须编写使用ObservableTransformer, SingleTransformer, .... 的不同函数？

Answer 1

reified我使用type创建了一个辅助方法：

inline fun <reified T> withSchedulers(): T {
    when (T::class) {
        ObservableTransformer::class  -> return ObservableTransformer<Unit, Unit> {
            it.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
        } as T
        SingleTransformer::class      -> return SingleTransformer<Unit, Unit> {
            it.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
        } as T
        CompletableTransformer::class -> return CompletableTransformer {
            it.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
        } as T
    }
    throw IllegalArgumentException("not a valid Transformer type")
}

例子 ：

    Observable.just("1", "2")
            .compose(withSchedulers<ObservableTransformer<String, String>>())
            .subscribe(System.out::println)

    Single.just(3)
            .compose(withSchedulers<SingleTransformer<Int, Int>>())
            .subscribe(Consumer { println(it) })

    Completable.defer { Completable.complete()  }
            .compose(withSchedulers<CompletableTransformer>())
            .subscribe { println("completed") }

输出 ：

1
2
3
completed

可能有其他方法可以做到这一点，但我想到了这一点。