java - 与 compose() 一起使用时，使用 RxJava/RxKotlin 2 自定义 Observable Transformer

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、Completable、Maybe 和 流动。因此，每次我想将它们与我的 withSchedulers() 函数结合起来时，我都必须将其转换为新类型(我不希望这样做)。

例如，

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, 可完成,...) ?或者我们必须编写使用 ObservableTransformer、SingleTransformer、.... 的不同函数？

Answer 1

我使用具体化类型创建了一个辅助方法:

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

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