1.结构
Rxlifecycle代码很少，也很好理解，来看核心类。
-
接口ActivityLifecycleProvider
RxFragmentActivity、RxAppCompatActivity、RxFragment等类所有的组件类皆实现这个借口
类RxLifecycle
2.详细分析
以RxAppCompatActivity入手来分析。
-
初始化一个BehaviorSubject，Subject因为它是一个Observer，它可以订阅一个或多个Observable；又因为它是一个Observable，它可以转发它收到(Observe)的数据，也可以发射新的数据。Subject的详细介绍:https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
private final BehaviorSubject<ActivityEvent>
lifecycleSubject = BehaviorSubject.create(); -
在activity每个生命的周期Subject发射相对应的事件。
@Override
@CallSuper
protected void onStart() {
super.onStart();
//发送start事件
lifecycleSubject.onNext(ActivityEvent.START);
}
....
分析RxLifecycle核心的方法：bindUntilActivityEvent
-
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)
);
}
};
}
- .share()操作符:
-
来看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结果。
-
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。
-
.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，则不发射任何数据。
-
source.takeUntil
如果combineLatest.onErrorReturn.takeFirst链返回true，则takeUntil操作符终止订阅，source Observable就回停止发射数据，执行Unsubscribe，流自动结束。
OVER!