反復された要素とサブスクライブの間の遅延

Question

固定されて所望の出力が得られます！

私は反復する必要があるアイテムのリストを持っており、各エレメント間には2秒待っています。すべてのアイテムが処理され、サブスクリプションが呼び出されると、もう2秒待っています。ここで

が私のコードです：

rx.Observable.from(new String[] {"Test1", "Test2", "Test3", "Test4"}) 
     .zipWith(rx.Observable.interval(2000, 1000,TimeUnit.MILLISECONDS), (a,b) -> a) 
     .subscribe(name -> { 
      System.out.println(name); 
     }, e -> { 
      System.out.println("ERROR" + e); 
     },() -> { 
      rx.Observable.timer(4000, TimeUnit.MILLISECONDS) 
         .subscribe(notUsed -> "THE END!"); 
     }); 


     try { 
      // Sleep so the program doesn't exit immediately 
      Thread.sleep(10000); 
     } 
     catch (Exception e) { 

     } 

それはかなりいい実行されますが、それが印刷さ購読で「間隔」実行時に「ENDを！」 3回！

TEST1 
TEST2 
TEST3 
TEST4 
THE END! 
THE END! 
THE END! 

これを改善する方法はありますか？

ありがとうございました！あなたの最初のストリーム（Test1-4）が完了したら

Answer 1

、あなたが遅延Observable.timer(2, TimeUnit.SECONDS).map(notUsed -> "The end")

コピーpastableコード

rx.Observable.from(new String[]{"Test1", "Test2", "Test3", "Test4"}) 
       .zipWith(rx.Observable.interval(2000, 1000, TimeUnit.MILLISECONDS), (a, b) -> a) 
       .concatWith(Observable.timer(2, TimeUnit.SECONDS).map(notUsed -> "The end")) 
       .subscribe(name -> { 
          System.out.println(name); 
         }, e -> { 
          System.out.println("ERROR" + e); 
         } 
       ); 


     try { 
      // Sleep so the program doesn't exit immediately 
      Thread.sleep(10000); 
     } catch (Exception e) { 

     } 

Answer 2

後に単一の値を発光ObservableでそれをCONCATことができますまあ、それはあなたが言うまさにませんそれ

time event 
------------------------------------- 
    0 start, wait 2000 ms initially 
2000 emit "Test1", wait 1000 ms 
3000 emit "Test2", wait 1000 ms 
4000 emit "Test3", wait 1000 ms 
5000 emit "Test4", emit the OnCompleted event, wait 2000 ms 
7000 print "THE END!", wait 1000 ms 
8000 print "THE END!", wait 1000 ms 
9000 print "THE END!", wait 1000 ms 
10000 main thread dies 

:-)行うには、私は正確に何をしたいかわからないが、私はあなただけのシングル「ENDを！」したいと推定します。あなたはこれが同様に1000のミリ秒遅れで表示されるようにしたい場合は、次の操作を実行する必要があります。

Observable.from(new String[] {"Test1", "Test2", "Test3", "Test4"}) 
    .zipWith(Observable.interval(2000, 1000,TimeUnit.MILLISECONDS), (a,b) -> a) 
    .concatWith(Observable.timer(1000, TimeUnit.MILLISECONDS).ignoreElements()) 
    .subscribe(
     System.out::println, 
     e -> System.out.println("ERROR" + e), 
     () -> System.out.println("THE END!)); 

あなたが最後に遅延したくない場合は、単に上記concatを削除します。

は何このconcatが行うことは、それが完了して放出される最初のシーケンスからOnCompleted一度Observerに加入する第一の配列（fromとzip）を待ちます。 2番目のシーケンスがサブスクライブされると、1000ミリ秒間待機し、要素を放出して完了します（intervalではなくtimerです）。要素（a.k.a. OnNext）を無視するため、OnCompletedイベントが残っているだけで、subscribeのコールバックがトリガーされます。