RxJava 2 flatMapCompletableで観測できません

Question

私はアイテムを放出し、それらをサーバにアップロードするオブザーバブルを持っています。ここで

はコードです：

repository 
      .getItems() 
      .doOnComplete(() -> Log.d(TAG, "No items left.")) 
      .flatMapCompletable(item -> 
        repository 
          .uploadItems(item) 
          .onErrorComplete() 
          .andThen(
            deleteTemporaryItem() 
              .onErrorComplete() 
          ) 
      ); 

getItemsメソッドは、アイテムを一つずつを発し、その後、完了、uploadItems方法は、サーバーにアップロードします。問題は、すべてのチェーンonCompleteイベントが正常に機能し、すべての購読者がこのイベントを取得して処理を進めるが、いくつかのアイテムがあり、すべてがonCompleteイベントがアップロードされたときに.doOnComplete（） - > Log.d（TAG、 "No items left。"））メソッドであり、すべてのサブスクライバはこのイベントを取得しません。私はonErrorCompleteを追加して、uploadItemsの後のすべてのメソッドが完了し、すべてのメソッドが完了していることをログで確認しましたが、repository.getItems（）のonCompleteイベントはすべてのサブスクライバに渡されません。

誰でもこの現象が発生する原因を突き止めてください。

ありがとうございます！

Answer 1

この例を見てください：

は、私は、各工程を経アイテムを渡すので、購読は処理された各項目に通知されます。処理パイプラインには、ファイルのアップロードと削除が含まれます。

実装を変更し、出力のログを投稿してください。

@Test 
void name() throws Exception { 
    Flowable<Integer> completed_work = Flowable.just(1, 2, 3) 
      .map(integer -> integer * 1000) 
      .flatMapSingle(integer -> 
        Completable.fromAction(() -> { 
         Thread.sleep(integer); 
         // do upload stuff here 
        }) 
          .doOnComplete(() -> System.out.println("Uploaded file ....")) 
          //.timeout(10, TimeUnit.SECONDS) 
          .retry(3) 
          .andThen(
            Completable.fromAction(() -> { 
             // do delete stuff... 
            }) 
              .retry(2) 
              //.timeout(10, TimeUnit.SECONDS) 
              .doOnComplete(() -> System.out.println("Deleted file ...")) 
          ) 
          .toSingle(() -> integer) 
      ) 
      .doOnComplete(() -> System.out.println("Completed work")); 


    completed_work.test() 
      .await() 
      .assertResult(1000, 2000, 3000); 
}