免费网站客服系统推广计划

案例代码：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

ok，那么下一篇就加入线程切换