Rxlifecycle(二):源码解析

Rxlifecycle(二):源码解析

1.结构

Rxlifecycle代码很少,也很好理解,来看核心类。

  • 接口ActivityLifecycleProvider

    RxFragmentActivity、RxAppCompatActivity、RxFragment等类所有的组件类皆实现这个借口

  • 类RxLifecycle

2.详细分析

以RxAppCompatActivity入手来分析。

  1. 初始化一个BehaviorSubject,Subject因为它是一个Observer,它可以订阅一个或多个Observable;又因为它是一个Observable,它可以转发它收到(Observe)的数据,也可以发射新的数据。Subject的详细介绍:https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
    private final BehaviorSubject<ActivityEvent> 
        lifecycleSubject = BehaviorSubject.create();
  2. 在activity每个生命的周期Subject发射相对应的事件。
    复制代码
    复制代码
    @Override
    @CallSuper
    protected void onStart() {
        super.onStart();
        //发送start事件
        lifecycleSubject.onNext(ActivityEvent.START);
    }
    ....
    复制代码
    复制代码

分析RxLifecycle核心的方法:bindUntilActivityEvent

  1. bindUntilActivityEvent也就是调用的bindUntilEvent
    复制代码
    复制代码
    private static <T, R> Observable.Transformer<T, T> bindUntilEvent(final Observable<R> lifecycle,        
    //返回 Transformer                                  final R event) {
    return new Observable.Transformer<T, T>() {
        @Override
        public Observable<T> call(Observable<T> source) {
            return source.takeUntil(
                lifecycle.takeFirst(new Func1<R, Boolean>() {
                    @Override
                    public Boolean call(R lifecycleEvent) {
                        return lifecycleEvent == event;
                    }
                })
            );
        }
    };
    }
    复制代码
    复制代码

     

这个方法接收两个参数,lifecycle就是activity里面的BehaviorSubject对象,event就是要我们设置的要在activity哪个生命周期取消订阅的ActivityEvent对象。

返回参数是Transformer,用来结合compose使用 Transformer相当于一个过滤器,Observable call(Observable source) 接收一个Observable然后经过处理再返回一个Observable

这个方法从里到外一层一层剥开:

复制代码
复制代码
lifecycle.takeFirst(new Func1<R, Boolean>() {
                    @Override
                    public Boolean call(R lifecycleEvent) {
                        return lifecycleEvent == event;
                    }
                })
复制代码
复制代码

如果lifecycleEvent == event结果为true,lifecycle既BehaviorSubject对象发射一个数据。

lifecycleEvent是BehaviorSubject发射的数据,既ActivityEvent对象,比如在onStart时候lifecycleSubject.onNext(ActivityEvent.START)发送的ActivityEvent.START。 event是传递进来的参数。

接着上看

return source.takeUntil()

lifecycle*核心的就是这个takeUntil。

source就是要调用compose的原始的Observable,就是例子中这个Observable

  Observable.interval(1, TimeUnit.SECONDS)
      .compose(this.<Long>bindUntilEvent(ActivityEvent.PAUSE))
      ...

来看takeUntil:

Returns an Observable that emits the items emitted by the source Observable until a second Observable emits an item.

如果lifecycle.takeFirst发射了一条数据,takeUntil就回触发,source Observable就回停止发射数据,执行Unsubscribe,流自动结束。

分析RxLifecycle核心的方法:bindUntilActivityEvent

上代码

复制代码
复制代码
private static <T, R> Observable.Transformer<T, T> bind(Observable<R> lifecycle,
                                                        final Func1<R, R> correspondingEvents) {
        ...
    // Make sure we're truly comparing a single stream to itself
    final Observable<R> sharedLifecycle = lifecycle.share();

    // Keep emitting from source until the corresponding event occurs in the lifecycle
    return new Observable.Transformer<T, T>() {
        @Override
        public Observable<T> call(Observable<T> source) {
            return source.takeUntil(
                Observable.combineLatest(
                    sharedLifecycle.take(1).map(correspondingEvents),
                    sharedLifecycle.skip(1),
                    new Func2<R, R, Boolean>() {
                        @Override
                        public Boolean call(R bindUntilEvent, R lifecycleEvent) {
                            return lifecycleEvent == bindUntilEvent;
                        }
                    })
                    .onErrorReturn(RESUME_FUNCTION)
                    .takeFirst(SHOULD_COMPLETE)
            );
        }
    };
}
复制代码
复制代码
  1. .share()操作符:
  2. 来看Observable.combineLatest,这个操作符接收三个参数。

    *个参数:取BehaviorSubject发射的数据中的*个,然后转换成对应的生命周期。例如在onStart()中调用了bindToLifecycle,take(1)后的数据是ActivityEvent.START,经过map(),返回ActivityEvent.STOP。

    第二个参数:从BehaviorSubject发射的数据中经过.skip(1)操作符,过滤掉*个数据。例如在onStart()中调用了bindToLifecycle,在后续的生命周期中会收到,ActivityEvent.RESUME、ActivityEvent.PAUSE、ActivityEvent.STOP、ActivityEvent.DESTROY

    第三个参数:作用是combineFunction,把前两个参数*近发射的数据按照规则进行合并。规则是比对两次事件是否相等,然后合并后数据返回Boolean结果。比如params2发射ActivityEvent.RESUME的时候,和params1发射的ActivityEvent.STOP进行比对,返回false结果;params2发射ActivityEvent.STOP的时候,和params1发射的ActivityEvent.STOP进行比对,返回true结果。

  3. onErrorReturn()
    复制代码
    复制代码
    private static final Func1<Throwable, Boolean> RESUME_FUNCTION = new Func1<Throwable, Boolean>() {
    @Override
    public Boolean call(Throwable throwable) {
        if (throwable instanceof OutsideLifecycleException) {
            return true;
        }
    
        Exceptions.propagate(throwable);
        return false;
    }
    };
    复制代码
    复制代码

    如果发生错误,判断是否是自定义错误类型 OutsideLifecycleException,如果是,则返回true,否则其他错误类型返回false。

  4. .takeFirst(SHOULD_COMPLETE)
    复制代码
    复制代码
    private static final Func1<Boolean, Boolean> SHOULD_COMPLETE = new Func1<Boolean, Boolean>() {
        @Override
        public Boolean call(Boolean shouldComplete) {
            return shouldComplete;
        }
    };
    复制代码
    复制代码

    返回*个结果是true的数据。如果combineLatest链中返回false,则不发射任何数据。

  5. source.takeUntil

    如果combineLatest.onErrorReturn.takeFirst链返回true,则takeUntil操作符终止订阅,source Observable就回停止发射数据,执行Unsubscribe,流自动结束。

OVER!