rxjava2 - 無限整数シーケンスジェネレータを作成する方法は？

Question

rxjava2を使用して、2つの無限整数ストリームシーケンスを作成し、ペアを作成して結果を出力するためにシーケンスを結合します。

私が理解できない理由のために、私のコードは、zip関数が実行されているときに無限ループを作成します。 zip関数は、無限のシーケンスが完了してからアイテムを放出するのを待っているようです。

以下の作業を行うにはどうすればよいですか？

例：

• SEQ1 = 1、2、3、...、無限
• SEQ2 = 5、10、15、...、無限
• ジッパー（SEQ1、 SEQ2 ）=（1,5）、（2,10）、（3,15）、...、

無限私のコードは次の通りである：

public Observable<Integer> intervalSequence(int start, int step) 
{ 
    return Observable.generate(()->start, 
    (s, emitter)->{ 
     int next = s+step; 
     emitter.onNext(next); 
     return next; 
    }); 
} 

public Observable<Pair<Integer, Integer>> pageRange(int pageSize){ 
    Observable<Integer> from = intervalSequence(1, pageSize); 
    Observable<Integer> to = intervalSequence(pageSize, pageSize); 
    Observable<Pair<Integer, Integer>> fromTo = Observable.zip(from, to, (f, t) -> { 
     return Pair.of(f, t); 
    }); 
    return fromTo; 
} 

@Test 
public void pageRangeTest() throws Exception { 
    //works 
    intervalSequence(1, 5).take(5).doOnNext(nmb -> System.out.println(nmb)).subscribe(); 
    //Doesn't work 
    pageRange(5).take(5).doOnNext(pair -> System.out.println("left: "+pair.getLeft()+" right: "+pair.getRight())).subscribe(); 
} 

pageRange(5).take(5).doOnNext(pair -> System.out.println("left: "+pair.getLeft()+" right: "+pair.getRight())).subscribe();は自分の値を印刷できますか？

ありがとうございました。

Answer 1

@akarnokdからのフィードバックに続いて、Observableの代わりにFlowableを使用すると、問題は解決します。 Flowableを使用すると、作業コードは