非同期で複数の API を並列で通信した結果を全て待機してから次の処理に進む方法が知りたいです

Question

複数 API を並列で通信させて、結果が全て揃うまで待機して次の処理に進む方法が知りたいです

表題の件を実現するために Android で RxJava を初めて使って処理を書いていますが、並列で通信に行くものの結果を待つことができず、解決策を探しています。

RxJava に詳しい方がいらっしゃいましたら、解決方法を教えていただけませんでしょうか。

よろしくお願いします。

ソースコード

class GetAllData() {
	final GetData g1 = new GetData();
	final GetData g2 = new GetData();

	Observabole<Boolean> task1 = Observable.create(new ObservableOnSubscribe() {
		@Override
		public void subscribe(ObservableEmitter emitter) throws Exception {
			g1.async(emitter);
		}
	}

	Observabole<Boolean> task2 = Observable.create(new ObservableOnSubscribe() {
		@Override
		public void subscribe(ObservableEmitter emitter) throws Exception {
			g2.async(emitter);
		}
	}

	Observable.zip(task1, task2, new BiFunction<Boolean, Boolean, Booelan>() {
		@Override
		public Boolean apply(Boolan result1, Boolean result2) throws Exception {
			log.d("TEST", "complete:"); ← 全ての処理が終わったら呼ばれると思っているが呼ばれない状態です
			return true;
		}
	}).subscribe();
}

class GetData {
	public void async(ObservableEmitter emitter) {
		this.emitter = emmiter;
		apiCall();    // 通信してデータを取得したら response() が呼ばれる
	}

	@Override
	public response(int code, String data) {
		if (data == null) {
			emitter.onError(new Throwable("error"));    // 取得失敗
		} else {
			emitter.onNext(true);    // データ取得成功
		}
	}

}

Comments

[xebme]

CountDownLatchを調べて追記してください。

[toumorokoshi]

CountDownLatch は知らなかったので参考にさせていただきます
貴重な情報ありがとうございました

Answer 1

ぱっと見ではRxJavaに関する実装に問題はなさそうです。GetDataのresponseがそもそも呼ばれていないかonErrorに入っているかのどちらかだと思うので、その辺りを確認してください。

あと1点指摘するとすれば、Observable.zipの第３引数は各結果の合成をする関数を渡すためのものなので、そこでログ出力をするのは適切ではないです。subscribeのonNextやonError等の引数（もしくは、doOnNextやdoOnError等のオペレータ）を使いましょう。

Comments

[toumorokoshi]

年末年始でお返事が遅くなりました
GetData の response が呼ばれていない状態でした
きちんと呼ばれるように修正したところ、意図した動作になりました
あと、ログ出力については適切な位置(onNext, onError)で行うように留意します
ありがとうございました