（4.6.22.3）来吧，是时候撸一份自己的RxJava框架啦：装饰器模式实现操作符顺序调用

操作符的原理： 适配器模式(装饰器模式)

3.1 数据结构链式结构的头插法，实现装饰器模式

3.2 头部调用

设计思路

装饰器模式，实现onObserver代理的通用集成，并持有上一个装换

图形实例

Publisher

.create()

.h1 // 创建observer类型的h1， header 和 last指针都指向该节点，h1.pre为null

.h2 // 创建observer类型的h2，h2.pre为null。设置h1.pre = h2， 同时 last后移

.h3// 创建observer类型的h3，h3.pre为null。设置h2.pre = h3， 同时 last后移

.bind(observerA)

.post();// 设置h3.pre = observerA， 并将header 传递给 生产者.call(header)方法中

此时，我们在call中调用 header.onSuccess(T)时，其实是 ：

先调用Pre=h1.Success{1. 完成h1的func1 2. 调用h1.Pre=h2.Success

{1. 完成h2的func1 2. 调用h2.Pre=h3.Success

{1. 完成h3的func1 2. 调用h3.Pre=onObserver.Success}}}

实践

1. 链式结构

1.1 抽象节点

/**
* 【观察者---消费者： 事件的消费者】 的控制器代理： 变换类通用代理
* 实现链式调用的数据结构
* Created by yhf on 16/12/18.
*/
public abstract class Change<I,O> implements OnObserver<I>,IObserverProxy<O> {
protected OnObserver<O> preChange; // 上个变换实例---指向上一个指针
protected Func1 changeImpl; // 转换代理 实例
protected IObserverProxy iObserverProxy;// 被代理的观察者们的 【实际】控制器代理
public Change(OnObserver<I> preChange) {
this.preChange = (OnObserver<O>) preChange;
iObserverProxy = (IObserverProxy<I>) preChange;
}
public Change() {
}

/**
* 调用 【实际 【观察者---消费者： 事件的消费者】】控制器 代理   向观察者容器中，增删查
*
*/
@Override
public void addObserver(OnObserver observer) {
iObserverProxy.addObserver(observer);
}

@Override
public void rmObserver(OnObserver observer) {
iObserverProxy.addObserver(observer);
}

@Override
public void clear() {
iObserverProxy.clear();
}

@Override
public Vector<OnObserver<O>> getObservers() {
return iObserverProxy.getObservers();
}

/**
* //调用 preChange 代理
*/
@Override
public void onError(Throwable throwable) {
preChange.onError(throwable);
}

@Override
public void onFinished() {
preChange.onFinished();
}

/**
* 完成链式结构
* @param preChange
*/
public void setPreChange(OnObserver<I> preChange){
this.preChange = (OnObserver<O>) preChange;
iObserverProxy = (IObserverProxy<I>) preChange;
}

public void setChangeImpl(Func1 changeImpl) {
this.changeImpl = changeImpl;
}
}

1.1.1 过滤操作的 链表节点

我们可以看到，它是先调用 自身的 过滤实例函数，如果成功了，才会调用前一个的

/**
* 【观察者---消费者： 事件的消费者】 的控制器代理： 过滤类同用代理
* 链式的数据结构的具体节点
* Created by yhf on 16/12/18.
*/
public class FilterChange<T> extends Change<T,T>{

/**
* //唯一  没有直接使用  实际代理控制器的地方
// !!! 根据preChange代理mapProxy.call(t)的返回值，作为参数，调用onSucces
* @param t
*/
@Override
public void onSuccess(T t) {
if ((Boolean) changeImpl.call(t))
preChange.onSuccess(t);
}
}

1.1.2 map 数据转换操作的 链表节点

/**
* 【观察者---消费者： 事件的消费者】 的控制器代理： 变换类同用代理
* 链式的数据结构的具体节点
* Created by yhf on 16/12/18.
*/
public class MapChange<I,O> extends Change<I,O>{

/**
* //唯一  没有直接使用  实际代理控制器的地方
// !!! 根据preChange代理mapProxy.call(t)的返回值，作为参数，调用onSucces
* @param i
*/
@Override
public void onSuccess(I i) {
preChange.onSuccess((O) changeImpl.call(i));
}
}

2. 图形实例的实现过程

private Change changeHead;
private Change changeLast;
private Publisher() {
...
changeHead = new Change() {
@Override
public void onSuccess(Object o) {
preChange.onSuccess(o);
}
};
changeLast = changeHead;
...
}

/**
* 实现链式的顺序调用方式,就是图示的h1.2.3操作
* 每个变换代理中持有上个变换代理实例，调用上个变换代理的变换函数
*   h1------pre----> h2------pre----->h3
*  head                              Last
*  其实就是 数据结构的链式结构，尾插法
* @param change  变换的实例对象
* @param changeProxy 变换的通用代理
* @param <I>
* @param <O>
* @return
*/
@Override
public <I, O> IPublisher<T> change(Func1<I, O> change, Change changeProxy) {
changeProxy.setChangeImpl(change);// 设置节点的自身操作函数
changeLast.setPreChange(changeProxy); // 设置节点前驱
changeLast = changeProxy; // last指针后移
return this;
}

@Override
public Future post() {
/**
*  h1------pre----> h2------pre----->h3---->onObserver
*  head                              Last
*  其实就是 数据结构的链式结构，尾插法
*/
changeLast.setPreChange(onObserver);
/**
*  h1------pre----> h2------pre----->h3---->onObserver
*  head                              Last
*  调用changeHead.onSuccess的过程就是：
*  先调用Pre=h1.Success{1. 完成h1的func1 2. 调用h1.Pre=h2.Success
*                                                        {1. 完成h2的func1 2. 调用h2.Pre=h3.Success
*                                                                                              {1. 完成h3的func1 2. 调用h3.Pre=onObserver.Success}}}
*/
onPublisher.call(changeHead); // 注入 消费者们的代理类
OnPublisherProxy proxy = (OnPublisherProxy) onPublisher;
return proxy.getFuture();
}