手写 RxJava ---- map 操作符

一、实现的思路

RxJava 提供了很多的变换操作符,将上游的数据转换成另一种数据，可以在传送数据流的过程中构造一个新的类，这个类即持有上游的 Observable 也持有下游的 Observer 和 变换的能力来实现

二、具体代码

1.创建 map 操作符

map 操作符本身持有 ObservableOnSubscribe 对象，这里将 ObservableOnSubscribe 交给一个 ObservableMap 让它持有了上游的能力。同时也持有了变换的能力

  /**
     * 自定义 map 操作符
     * map 操作符会拿到 SelfObservableMap 所持有的 Observable 对象
     * 交给一个 ObservableMap 对象 然后替换一个新的（持有 ObservableMap 这个 ObservableOnSubscribe 的引用，已经一个转换的函数）
     * 即拥有控制上一层的能力 也拥有控制下一层的能力
     * <p>
     * 变换操作符只考虑上一层的类型 变换成新的类型后 给到下一层
     */
    public <R> SelfObserverable<R> map(CFunction<? super T, R> function) {
        ObservableMap observableMap = new ObservableMap(observable, function);
        // 这里会把 source 替换成 ObservableMap
        // 在 subscribe 时实际调用的 observable map 所持有的 source 的 subscribe 方案
        return new SelfObserverable<>(observableMap);
    }

2.ObservableMap 同时提供了 subscribe 函数，让它持有了下游的能力

/**
 * 为 Map 专门定义的 ObservableOnSubscribe
 * T 接收的类型 R 返回的类型
 *
 */
public class ObservableMap<T,R> implements ObservableOnSubscribe<R> {
    // 这里持有的是第一次通过 create 函数创建持有的 ObservableOnSubscribe 对象
    private ObservableOnSubscribe observable;
    private CFunction<? super T, ? extends R> function;

    public ObservableMap(ObservableOnSubscribe source, CFunction<? super T, ? extends R> function) {
        this.observable = source;
        this.function = function;
    }

    /**
     * observableEmitter 是外层通过 subscribe 函数传递进来的 Observer 类
     * @param observer
     */
    @Override
    public void subscribe(Observer<? super R> observer) {
        // SelfObserverable 调用 subscribe 函数时 实际上调用的是
        MapObserver mapObserver = new MapObserver(observer,  function);
        // observable 是通过 create 或者 just 传递进来的 observable，现在交给了它一个 MapObserver
        /**
         *     observable ----> create 的 new ObservableOnSubscribe
         *     // 使用 Map 操作符
         *     SelfObserverable.create(new ObservableOnSubscribe<Integer>() {
         *      @Override
         *      public void subscribe(Observer<? super Integer> observableEmitter) {
         *            observableEmitter.onNext(1);
         *            observableEmitter.onComplete();
         *         }
         *     })
         *
         *    这个 observable.subscribe 调用的就是 ObservableOnSubscribe 的 subscribe 函数的方法。
         *    会触发 MapObserver 类的 onNext 和 onComplete
         */
        observable.subscribe(mapObserver);
    }

3.在 subscribe 时又对 observer 做了进一步的包装,让被观察者 ObservableOnSubscribe 持有了 MapObserver 这个新的观察者类。在调用 onNext 时首先调用 CFunction 函数进行一次数据转换

 /**
     * 对 Observer 的包装
     *
     * @param <T>
     */
    class MapObserver<T> implements Observer<T> {
        Observer<? super R> observableEmitter;
        ObservableOnSubscribe source;
        CFunction<? super T, ? extends R> function;

        /**
         *
         * @param observableEmitter  通过 subscribe 传递进来的观察者
         * @param function function 是负责转换的函数
         */
        public MapObserver(Observer<? super R> observableEmitter,
                           CFunction<? super T, ? extends R> function) {
            this.observableEmitter = observableEmitter;
            this.function = function;
        }

        @Override
        public void onSubscirbe() {
            observableEmitter.onSubscirbe();
        }

        @Override
        public void onNext(T value) {
            // 将转换后的值交给 onNext
            R next = function.apply(value);
            observableEmitter.onNext(next);
        }

        @Override
        public void onComplete() {
            observableEmitter.onComplete();
        }

        @Override
        public void onError(Throwable throwable) {
            observableEmitter.onError(throwable);
        }
    }

四、流程示例

RxJava-2.jpg

五、使用示例

  // 使用 Map 操作符
        SelfObserverable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(Observer<? super Integer> observableEmitter) {
                observableEmitter.onNext(1);
                observableEmitter.onComplete();
            }
        }).map(new CFunction<Integer, String>() {

            @Override
            public String apply(Integer integer) {
                return "我被转换啦";
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscirbe() {
                Log.d("TAG", "on subscribe with map");
            }

            @Override
            public void onNext(String value) {
                Log.d("TAG", "我被转换啦");
            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable throwable) {

            }
        });