案例代碼：https://gitee.com/bobidali/lite-rx-java/commit/292e9227a5491f7ec6a07f395292ef8e6ff69290
RxJava 的調用第一步是封裝了觀察者接受了數據的處理，進一步就是使用 map 將數據操作傳遞給上下游
1、類似Observer.create 創建一個簡單的觀察者
abstract class Single<T> : SingleSource<T> {companion object {@JvmStaticfun <T> just(item: T): Single<T> {return RxJavaPlugins.onAssembly(SingleJust(item))}}override fun subscribe(observer: SingleObserver<in T>) {Objects.requireNonNull(observer, "observer is null")try {var obsv = observerobsv = RxJavaPlugins.onSubscribe(this, obsv)subscribeActual(obsv)} catch (e: NullPointerException) {throw e} catch (e: Throwable) {throw NullPointerException("subscribeActual failed").initCause(e)}}protected abstract fun subscribeActual(observer: SingleObserver<in T>)
}
class SingleJust<T>(private val value: T) : Single<T>() {override fun subscribeActual(observer: SingleObserver<in T>) {observer.onSubscribe(Disposable.disposed())observer.onSuccess(value)}
}
2、在SingleObserver 中加入map 方法實現數據的承上啟下
fun <R> map(mapper: com.example.literj.kt.internal.functions.Function<in T, out R>): Single<R> {Objects.requireNonNull(mapper, "mapper is null")return RxJavaPlugins.onAssembly(SingleMap(this, mapper))}
class SingleMap<T, R>(private val source: SingleSource<out T>,private val mapper: Function<in T, out R>
) : Single<R>() {override fun subscribeActual(observer: SingleObserver<in R>) {source.subscribe(MapSingleObserver(observer, mapper))}internal class MapSingleObserver<T, R>(private val t: SingleObserver<in R>,private val mapper: Function<in T, out R>) : SingleObserver<T> {override fun onSuccess(t: T) {val v: Rtry {v = Objects.requireNonNull(mapper.apply(t),"The mapper function returned a null value.")} catch (e: Throwable) {onError(e)return}this.t.onSuccess(v)}override fun onSubscribe(d: Disposable) {t.onSubscribe(d)}override fun onError(e: Throwable) {t.onError(e)}}
}
3、鐺鐺，完成，看看調用案例
public static void sample3() {String tag = "sample3";Log.d(TAG + " " + tag, "start");Single.just(777).map(data1 -> 777).map(data2 -> 888).subscribe(getSingleObserver(tag));}
2023-08-13 11:58:23.703 21522-21522/com.example.literj D/RxJavaTestUtil sample3: start
2023-08-13 11:58:23.713 21522-21522/com.example.literj D/RxJavaTestUtil sample3: onSubscribe
2023-08-13 11:58:23.713 21522-21522/com.example.literj D/RxJavaTestUtil sample3: onSuccess = 888
