：

Question

を放射されたX項目よりも少ない場合、私はOBS1、obs2、obs3、...、

それらのそれぞれ（のMongoDBデータベースからの）項目の数を発することができる観測のリストを持っているスイッチは、私が興味を持っています最初のN個のアイテムにのみ表示されます。私は、必要に応じてオブザーバブルのクエリが実行されるようにしたいと思います。つまり、たとえばobs1がNより大きい値を生成した場合、obs2の後のクエリは実行されません。

私がconcatを使用すると、Observable（obs1、obs2、obs3、...）。concat、all Nアイテム未満である：クエリはMongoDBの基本的

で並列に実行することができ、私はX

..... obs1.switchIfX（obs2）.switchIfX（obs3）のような操作を探しています現在の観測可能なものから放出される。

どのように私はこの要件をrxscalaスタイルで実装できますか？

Answer 1

あなたが（未テスト）このような何かを試みることができる：

Observable.just(obs1, obs2, obs3).flatten(maxConcurrent=1).take(N) 

maxConcurrent引数はflattenは一度に観察できるものに加入し、そしてNアイテムが放出された後、takeは上流から退会することが保証さその時点でobs2またはobs3がまだ購読されていない場合、それらは決して実行されません。

Answer 2

あなたは、リストにソースからアイテムを収集flatMapオペレータにそのサイズをチェックして、長さが十分でない場合は、別のソースに切り替えることができます：

@Test 
public void test() { 
    Observable.range(1, 5) 
    .compose(switchIfFewer(Observable.range(1, 8), 10)) 
    .compose(switchIfFewer(Observable.range(1, 15), 10)) 
    .test() 
    .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); 
} 

static <T> ObservableTransformer<T, T> switchIfFewer(Observable<T> other, int n) { 
    return o -> { 
     return o.toList() 
     .flatMapObservable(list -> { 
      if (list.size() < n) { 
       return other; 
      } 
      return Observable.fromIterable(list); 
     }); 
    }; 
} 

あなたが取得したくない場合N個以上のアイテムの場合は、代わりにo.take(n).toList()と指定できます。