RxJava使用（四）变换

RxJava 提供了对事件序列进行变换的支持； 所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。

不仅可以针对事件对象，还可以针对整个事件队列。

变换部分主要来自《给Android 开发者的 RxJava 详解》

1. 事件对象变换-map()

map():事件对象的直接变换；它是 RxJava最常用的变换；可以将Observable深入的对象1转换为对象2发送给Subscriber。

基本用法如下：

Observable.just(R.drawable.t)
.map(new Func1<Integer, Drawable>() {
@Override
public Drawable call(Integer integer) {
return getResources().getDrawable(integer);
}
})
.subscribe(new Action1<Drawable>() {
@Override
public void call(Drawable drawable) {
showImg(drawable);
}
});

2. 事件序列变换-flatMap()

flatMap() 也和 map() 相同，也是把传入的参数转化之后返回另一个对象；和 map() 不同的是， flatMap() 中返回的是个 Observable 对象，并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。

flatMap() 的原理：

1. 使用传入的事件对象创建一个 Observable 对象；

2. 并不发送这个 Observable, 而是将它激活，于是它开始发送事件；

3. 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable ，而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。

这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。

扩展：由于可以在嵌套的Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。

flatmap()基本用法如下，定义了一个Book类型，包含名称和章节列表；通过flatmap()打印出全部的章节列表。

Ø Book.java

public class Book {
public String name;
public List<String> chapterList = new ArrayList<String>();

public void addChapter(String chapter) {
chapterList.add(chapter);
}
}

Ø flagmap()使用

Book[] books = getBookList(5);
Observable.from(books)
.flatMap(new Func1<Book, Observable<String>>() {
@Override
public Observable<String> call(Book book) {
return Observable.from(book.chapterList);
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.v(TAG, s);
}
});

3. 变换的原理：lift()

所有的变换功能可能有所不同，实质上都是针对事件序列的处理和再发送。而在 RxJava 的内部，它们是基于同一个基础的变换方法： lift(Operator)。

lift()实现的源码：

/**
* Lifts a function to the current Observable and returns a new Observable that when subscribed to will pass
* the values of the current Observable through the Operator function.
* 
* In other words, this allows chaining Observers together on an Observable for acting on the values within
* the Observable.
* {@code
* observable.map(...).filter(...).take(5).lift(new OperatorA()).lift(new OperatorB(...)).subscribe()
* }
* 
* If the operator you are creating is designed to act on the individual items emitted by a source
* Observable, use {@code lift}. If your operator is designed to transform the source Observable as a whole
* (for instance, by applying a particular set of existing RxJava operators to it) use {@link #compose}.
* <dl>
* <dt>Scheduler:</dt>
* <dd>{@code lift} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param operator the Operator that implements the Observable-operating function to be applied to the source
* Observable
* @return an Observable that is the result of applying the lifted Operator to the source Observable
* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators">RxJava wiki: Implementing Your Own Operators</a>
*/
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return new Observable<R>(new OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = hook.onLift(operator).call(o);
try {
// new Subscriber created and being subscribed with so 'onStart' it
st.onStart();
onSubscribe.call(st);
} catch (Throwable e) {
// localized capture of errors rather than it skipping all operators
// and ending up in the try/catch of the subscribe method which then
// prevents onErrorResumeNext and other similar approaches to error handling
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// if the lift function failed all we can do is pass the error to the final Subscriber
// as we don't have the operator available to us
o.onError(e);
}
}
});
}

通过lift()源码可以发现，生成了一个新的Observable（new）；并且在新的Observable（new）的OnSubscribe（new）的call()回调方法中，创建了一个新的Subscriber（new），该新的Subscriber（最终目标）使用了最终调用的Subscriber的代理；再使用原来的Observable（old）的onSubscribe（old）来调用新的Subscriber（new）。

图示lift()变换过程-（图片来自《给Android 开发者的 RxJava 详解》）

`lift()` 实现过程

1) lift() 创建一个新Observable ，此时加上之前的原始 Observable，已经有两个 Observable 了；

2) 而同样地，新 Observable 里的新 OnSubscribe 加上之前的原始 Observable 中的原始 OnSubscribe，也就有了两个 OnSubscribe；

3) 当用户调用经过 lift() 后的 Observable 的 subscribe() 的时候，使用的是 lift() 所返回的新的 Observable ，于是它所触发的 onSubscribe.call(subscriber)，也是用的新 Observable 中的新OnSubscribe，即在 lift() 中生成的那个 OnSubscribe；

4) 而这个新 OnSubscribe 的 call() 方法中的 onSubscribe ，就是指的原始 Observable 中的原始 OnSubscribe ，在这个 call() 方法里，新 OnSubscribe 利用 operator.call(subscriber) 生成了一个新的 Subscriber（Operator 就是在这里，通过自己的call() 方法将新 Subscriber 和原始 Subscriber 进行关联，并插入自己的『变换』代码以实现变换），然后利用这个新 Subscriber 向原始 Observable 进行订阅。

lift() 过程，有点像一种代理机制，通过事件拦截和处理实现事件序列的变换。

在 Observable 执行了lift(Operator) 方法之后，会返回一个新的 Observable，这个新的 Observable 会像一个代理一样，负责接收原始的 Observable 发出的事件，并在处理后发送给 Subscriber。

4. 自定义Observable.Operator，直接调用lift()

下面实例实现的功能跟前面flatMap()实现的功能一致：

Observable.from(getBookList(4))
.lift(new Observable.Operator<String, Book>() {
@Override
public Subscriber<? super Book> call(final Subscriber<? super String> subscriber) {
return new Subscriber<Book>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(Book book) {
for(String chapter : book.chapterList) {
subscriber.onNext(chapter);
}
}
};
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.v(TAG, s);
}
});

5. compose(Transformer)变换

compose(Transformer)变换，针对Observable 自身进行变换。假设在程序中有多个 Observable ，并且他们都需要应用一组相同的 lift() 变换，可以使用compose方式处理，而不是每个都调用一次相同的变换过程

实例代码，如需要将Book转换得到对应的全部章节，两个Observable对章节数据有不同的处理方式：

final Observable.Transformer bookTtransformer = new Observable.Transformer<Book, String>() {
@Override
public Observable<String> call(Observable<Book> bookObservable) {
return bookObservable.flatMap(new Func1<Book, Observable<String>>() {
@Override
public Observable<String> call(Book book) {
return Observable.from(book.chapterList);
}
});
}
};

Observable.from(getBookList(4))
.compose(bookTtransformer)
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
//处理方式1
Log.v(TAG, s);
}
});

Observable.from(getBookList(2))
.compose(bookTtransformer)
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
//处理方式2
Log.v(TAG, s);
}
});