java - Rx Java Observable 执行直到某些条件

Question

我正在尝试找到一种方法来执行 observable，直到满足某些条件。

考虑以下示例:

 myDelayedObservable = createListenerObserver();
    public Observable<Boolean> createListenerObserver() {

      // The part I am looking for
    }

    ViewTreeObserver.OnGlobalLayoutListener listenerLayout = new ViewTreeObserver.OnGlobalLayoutListener() {
        @Override
        public void onGlobalLayout() {
           myDelayedObservable.onCompleted();
         getTargetView().getViewTreeObserver().removeGlobalOnLayoutListener(this);

        }
    };

    public void performMultipleRequests() {

        Observable<Boolean> longRunningTask = Observable.zip(oneRequest, anotherRequest, myDelayedObservable,...);

    }

所以想法是运行多个请求，例如下载请求，连同使用 zip 的 myDelayedObservable，所以 longRunningTask 仅在以下情况下完成所有请求加上监听器(在我的例子中是 View 完成的布局)都已完成。

但问题是，我找不到为监听器创建 Observable 的正确方法。它就像一个障碍，所以伪代码

while(!viewIsLaidOut) {
  // just wait
}
observable.complete();
// After that `longRunningTask` should be completed

请建议实现此目的的正确方法，我考虑过Future、Callable，但这对我来说似乎不是最佳解决方案。

Answer 1

1) 你需要map所有可观察到的类型相同，例如。 Observable<Boolean> ，所以你可以合并它们:

observable1.map(String s -> "...".equals(s))
observable2.map(Integer i -> i > 0 && i < 100)
observable3.map(MyClass m -> true)
...

2) 使用Observable.merge()将它们全部合并为单个流。 使用zip为此，仅当所有可观察对象发出相同数量的项目时才有效，否则它将在第一个完成时立即完成，而无需等待其余的。

Observable<Boolean> allInOne = Observable.merge(observable1, observable2, ...);

3) myDelayedObservable只是那些应保持allInOne的可观察量之一不完整，直到一些听众回电。使用 Subject为此目的:

Subject<Boolean> myDelayedObservable = PublishSubject.create();

4) 当你的监听器准备好后，调用 myDelayedObservable.onComplete() .

5) 订阅 allInOne并在完成时使用react:

allInOne.subscribe(b -> { ... }, e -> { ... },
    () -> { ... go ahead with your next task ... });