RxJava - 等待Observable <List <Service >>中的所有服务'Observable <Boolean>

Question

我正在尝试建立一个监督是否所有服务Obsevable<List>都已准备就绪的经理.

代码的一般概念看起来类似于:

public class ServiceStuff {

    public interface Service {
        String getName();

        Observable<Boolean> monitorReady();
    }

    private BehaviorSubject<List<Service>> services = BehaviorSubject.createDefault(Collections.emptyList());

    ServiceStuff(List<Service> list) {
        services.onNext(list);
    }

    public void addService(Service service) {
        List<Service> newList = services.getValue();
        newList.add(service);
        services.onNext(newList);
    }

    public void removeService(Service service) {
        List<Service> newList = services.getValue();
        newList.remove(service);
        if (newList.remove(service)) {
            services.onNext(newList);
        }
    }

    public Observable<List<Service>> monitorServices() {
        return services.observeOn(AndroidSchedulers.mainThread());
    }

    public Observable<Boolean> monitorServicesReady() {
        return [If monitorReady() for all services return true, emit true, else false]
    }

}

如何在monitorServicesReady()不阻止UI线程的情况下实现该方法？应该可以在将任何服务添加到列表之前开始监视.

---

更新:

我现在尝试通过一些修改来实现解决方案cyroxis.blockingIterable()然而,ANR是应用程序,因为它永远不知道monitorServices()流何时结束.

public Observable<Boolean> monitorServicesReady() {
    Iterable<Observable<Boolean>> sources = monitorServices().flatMapIterable(x -> x).map(Service::monitorReady).blockingIterable();
    return Observable.combineLatest(sources, this::all);
}

private boolean all(Object[] values) {
    boolean result = true;
    for (boolean val : (Boolean[]) values) {
        result &= val;
    }
    return result;
}

Answer 1

你必须让你的服务也落实Observable。然后您可以使用操作符，当所有服务准备就绪时，操作符Merge也会被触发。Observable