複数のオブザーバブルが完了した後で何かをする

Question

私はAndroid上でRXJavaを使用していて、複数のAPIコールを連鎖しようとしており、両方のAPIコールが終了した後で何かを行います。私のAPIは、提供されているコードサンプルに似ています。基本的にAPI呼び出しを行い、onNextのDBに各レコードを書き込み、すべてのレコードが書き込まれた後に、いくつかのキャッシュを更新します。私はこれらの呼び出しを両方とも非同期的に起動し、両方がonCompletedを呼び出した後に何か他のことをしたいと思います。これを行うRXの正しい方法は何ですか？私は、私は別のストリームを結びつける必要はないので、私はジップが必要とは思わない。私はおそらくマージすると思っていたが、私の2つのAPI呼び出しは異なるタイプのObservableを返す。私にお知らせください。ありがとう。

getUsers() 
      .subscribeOn(Schedulers.io()) 
      .observeOn(Schedulers.io()) 
      .flatMap(Observable::from) 
      .subscribe(new Subscriber<User>() { 
       @Override 
       public void onCompleted() { 
        updateUserCache(); 
       } 

       @Override 
       public void onError(Throwable e) { 
        Log.e(TAG, "Error loading users", e); 
       } 

       @Override 
       public void onNext(User user) { 
        insertUserToDB(user); 
       } 
      }); 

    getLocations() 
      .subscribeOn(Schedulers.io()) 
      .observeOn(Schedulers.io()) 
      .flatMap(Observable::from) 
      .subscribe(new Subscriber<Location>() { 
       @Override 
       public void onCompleted() { 
        updateLocationCache(); 
       } 

       @Override 
       public void onError(Throwable e) { 
        Log.e(TAG, "Error loading Locations", e); 
       } 

       @Override 
       public void onNext(Location location) { 
        insertLocationToDB(location); 
       } 
      }); 

Answer 1

あなたは正しく考えています。 zip演算子を使用する必要があります。

あなたのすべての関数は、呼び出しを行い、データベースに書き込んで、必要なものをすべて実行する必要があります。 Theat zip出力機能の違い：呼び出されるとObservableのすべてが正常に完了したことを確認することができます。

がObservableのリストを作成します。

List<Observable<?>> observableList = new ArrayList<>(); 
observableList.add(
     getUsers() 
      .subscribeOn(Schedulers.io()) 
      .observeOn(Schedulers.io()) 
      .flatMap(Observable::from) 
      .insertUserToDB(user) 
      .toList()); 

observableList.add(
     getLocations() 
      .subscribeOn(Schedulers.io()) 
      .observeOn(Schedulers.io()) 
      .flatMap(Observable::from) 
      .insertLocationToDB(location) 
      .toList()); 

その後zip alll Observable年代：

Observable.zip(observableList, new FuncN<Object, Observable<?>>() { 
    @Override 
    public Observable<?> call(Object... args) { 
     return Observable.empty(); 
    } 
}).subscribe(new Subscriber<Object>() { 
    @Override 
    public void onCompleted() { 
     updateUserCache(); 
     updateLocationCache(); 
    } 

    @Override 
    public void onError(Throwable e) { 

    } 

    @Override 
    public void onNext(Object o) { 

    } 
}); 

をこれは擬似コードですが、私はあなたが考えを理解してほしいです。

Answer 2

.zipファイル（）それを

を行うための適切な方法であるあなたはレトロフィットは、誰かがそれを必要とする場合には

Answer 3

かかわらシングル代わりの観測を返すようにしたいことがあり、ここで私はR.に基づいて使用されるコードですザゴフスキーの提案：

List<Observable<?>> observableList = new ArrayList<>(); 
    observableList.add(
      getUsers() 
       .subscribeOn(Schedulers.io()) 
       .observeOn(Schedulers.io()) 
       .flatMap(Observable::from) 
       .doOnNext(user->insertUser(user)) 
       .toList() 
    ); 
    observableList.add(
      getLocations() 
        .subscribeOn(Schedulers.io()) 
        .observeOn(Schedulers.io()) 
        .flatMap(Observable::from) 
        .doOnNext(location->insertLocation(location)) 
        .toList() 
    ); 

    Observable.zip(observableList, new FuncN<Object>() { 
     @Override 
     public Observable<?> call(Object...args) { 
      return Observable.empty(); 
     }).subscribe(new Subscriber<Object>() { 
      @Override 
      public void onCompleted() { 
       updateUserCache(); 
       updateLocationCache(); 
      } 

      @Override 
      public void onError(Throwable e) { 

      } 

      @Override 
      public void onNext(Object o) { 

      } 
     });