RX: Пропустить Zipped Observables параллельно?

Итак, я играю с RX (действительно круто), и я конвертировал свой api, который обращается к базе данных sqlite в Android, чтобы вернуть наблюдаемые.

Поэтому, естественно, одна из проблем, которые я начал пытаться решить, заключается в следующем: "Что, если я хочу сделать 3 вызова API, получить результаты, а затем выполнить некоторую обработку, как только они закончатся?"

Мне потребовался час или 2, но в итоге я нашел Функцию Zip, и это помогает мне удобно:

    Observable<Integer> one = getNumberedObservable(1);
    Observable<Integer> two = getNumberedObservable(2);
    Observable<Integer> three = getNumberedObservable(3);

    Observable.zip(one, two, three, new Func3<Integer, Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer arg0, Integer arg1, Integer arg2) {
            System.out.println("Zip0: " + arg0);
            System.out.println("Zip1: " + arg1);
            System.out.println("Zip2: " + arg2);
            return arg0 + arg1 + arg2;
        }
    }).subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer arg0) {
            System.out.println("Zipped Result: " + arg0);
        }
    });

public static Observable<Integer> getNumberedObservable(final int value) {
    return Observable.create(new OnSubscribeFunc<Integer>() {
        @Override
        public Subscription onSubscribe(Observer<? super Integer> observer) {
            observer.onNext(value);
            observer.onCompleted();
            return Subscriptions.empty();
        }
    });
}

Отлично! Так что круто.

Итак, когда я застегиваю 3 наблюдаемых, они запускаются серийно. Что делать, если я хочу, чтобы все они работали параллельно одновременно, поэтому я получаю результаты быстрее? Я играл с несколькими вещами и даже пытался прочитать некоторые из оригинальных материалов RX, которые люди написали в С#. Я уверен, что есть простой ответ. Может кто-то указать мне верное направление? Каков правильный способ сделать это?

Ответы

public static Integer getNumber(final int value) { return value; }
public static Observable<Integer> getNumberedObservable(final int value) {
    return rx.util.functions.toAsync(new Func<Integer,Integer>() {
        @Override
        public Integer call(Integer value) { return getNumber(value); }
    });
};