介绍

装饰器模式（Decorator Pattern）又叫包装模式（Wrapper Pattern），允许向一个现有的对象添加新的功能，同时又不改变其结构。

这种类型的设计模式属于结构型模式，它是作为现有的类的一个包装。

这种模式创建了一个装饰类，用来包装原有的类，并在保持类方法签名完整性的前提下，提供了额外的功能。

ObservableCreate

将 RxJava 中的设计模式（一）观察者模式 一文中的 Observable 调整为下面的结构。

public abstract class Observable<T> {

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        return new ObservableCreate<>(source);
    }

    protected abstract void subscribeActual(Observer<T> observer);

    public void subscribe(Observer<T> observer) {
        if (observer == null) {
            throw new IllegalArgumentException("observer must not be null");
        }
        subscribeActual(observer);
    }
}

public class ObservableCreate<T> extends Observable<T>{
    private ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        Emitter<T> emitter = new Emitter<>(observer);
        this.source.subscribe(emitter);
    }
}

实现 ObservableMap

package org.company.rxjava.pattern;

import java.util.function.Function;

public class ObservableMap<U, D> extends Observable<D> {
    private Observable<U> upstream;
    private Function<U, D> mapper;

    public ObservableMap(Observable<U> upstream, Function<U, D> mapper) {
        this.upstream = upstream;
        this.mapper = mapper;
    }

    @Override
    protected void subscribeActual(Observer<D> observer) {
        MapObserver<U, D> mapObserver = new MapObserver<>(observer, this.mapper);
        this.upstream.subscribe(mapObserver);
    }

    public static class MapObserver<U, D> implements Observer<U> {
        private Observer<D> downstream;
        private Function<U, D> mapper;

        public MapObserver(Observer<D> downstream, Function<U, D> mapper) {
            this.downstream = downstream;
            this.mapper = mapper;
        }

        @Override
        public void onNext(U r) {
            D value = this.mapper.apply(r);
            this.downstream.onNext(value);
        }
    }
}

在 Observable 中添加如下方法：

    public final <R> Observable<R> map(Function<T, R> mapper) {
        return new ObservableMap<>(this, mapper);
    }

测试代码：

@Slf4j
public class Main {

    public static void main(String[] args) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(Emitter<Integer> emitter) {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onNext(3);
                    }
                })
                .map(v -> v + 10) // 11 12 13
                .map(v -> v * 2) // 22 24 26
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onNext(Integer o) {
                        log.info("onNext {}", o);
                    }
                });
    }
}

输出：

11:26:14.547 [main] INFO org.company.rxjava.pattern.Main - onNext 22
11:26:14.557 [main] INFO org.company.rxjava.pattern.Main - onNext 24
11:26:14.557 [main] INFO org.company.rxjava.pattern.Main - onNext 26

ObservableCreate 和 ObservableMap 职责和数据流向如下图所示：

参考

• RxJava So Easy! 带你从0手撸一个RxJava
• Rxjava中的设计模式
• RxJava 源码学习手写框架二 (前传) - 装饰者模式
• 装饰器模式