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.  * <p>  * In other words, this allows chaining Observers together on an Observable for acting on the values within  * the Observable.  * <p> {@code  * observable.map(...).filter(...).take(5).lift(new OperatorA()).lift(new OperatorB(...)).subscribe()  * }  * <p>  * 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><b>Scheduler:</b></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);           }       });