io.reactivex.Flowable#takeUntil ( )源码实例Demo

/**
 * 绑定一个 Activity、Fragment 的生命周期，自动释放资源
 * <br/>
 * 例如：网络请求时绑定{@link ActivityEvent#STOP} 或 {@link FragmentEvent#STOP},
 * onStop()时会自动取消网络请求.
 *
 * @param event 事件类型
 * @see ActivityEvent
 * @see FragmentEvent
 */
@SuppressWarnings("SpellCheckingInspection")
@Override
public <Type> FlowableTransformer<Type, Type> bindEventWithFlowable(final int event) {
    final Flowable<Integer> observable = mBehaviorSubject
            .toFlowable(BackpressureStrategy.LATEST)
            .filter(new Predicate<Integer>() {
                @Override
                public boolean test(Integer integer) throws Exception {
                    return integer == event;
                }
            })
            .take(1);

    return new FlowableTransformer<Type, Type>() {
        @Override
        public Publisher<Type> apply(@io.reactivex.annotations.NonNull Flowable<Type> upstream) {
            return upstream.takeUntil(observable);
        }
    };
}
 

/**
 * 绑定一个 Activity、Fragment 的生命周期，自动释放资源
 * <br/>
 * 例如：网络请求时绑定{@link ActivityEvent#STOP} 或 {@link FragmentEvent#STOP},
 * onStop()时会自动取消网络请求.
 *
 * @param event 事件类型
 * @see ActivityEvent
 * @see FragmentEvent
 */
@SuppressWarnings("SpellCheckingInspection")
@Override
public <Type> FlowableTransformer<Type, Type> bindEventWithFlowable(final int event) {
    final Flowable<Integer> observable = mBehaviorSubject
            .toFlowable(BackpressureStrategy.LATEST)
            .filter(new Predicate<Integer>() {
                @Override
                public boolean test(Integer integer) throws Exception {
                    return integer == event;
                }
            })
            .take(1);

    return new FlowableTransformer<Type, Type>() {
        @Override
        public Publisher<Type> apply(@io.reactivex.annotations.NonNull Flowable<Type> upstream) {
            return upstream.takeUntil(observable);
        }
    };
}
 

@Override
public Publisher<T> apply(Flowable<T> upstream) {
    return upstream.takeUntil(mObservable.toFlowable(BackpressureStrategy.LATEST));
}
 

@Override
public Publisher<T> apply(Flowable<T> upstream) {
    return upstream.takeUntil(mObservable.toFlowable(BackpressureStrategy.LATEST));
}
 

@Override
public Publisher<T> apply(Flowable<T> upstream) {
    return upstream.takeUntil(observable.toFlowable(BackpressureStrategy.LATEST));
}