反応性の拡張機能：分割オブジェクト

Question

を開始することにより、私はint型の観察可能としましょう：obs = 1 :: 2 :: 3 :: 1 :: 3 :: 1 :: 1 :: 2を、以下のように私は、1によってシリーズの開始を蓄積したい：

acc_obs = 1 + 2 + 3 :: 1 + 3 :: 1 :: 1 + 2 = 6 :: 4 :: 1 :: 3。

どのように反応拡張機能を使用して対応できますか？

私はobs.filter(i -> i == 1)を使用して考えたとbuffer(bufferClosingSelector)のいくつかの種類であることを使用しますが、二つの問題があるように思える：

• を私bufferClosingSelector「閉じ」場合、それは冷たい観測
• では動作しません。バッファが1を見つけるたびに、私は1のオフセットを持ち、1 :: 2 + 3 + 1 :: 3 + 1 :: 1のようなものを放射します。

Answer 1

私はSubjectとWindowで解決しましたが、それは良く見えません。どうぞご覧ください：

@Test 
public void name() throws Exception { 
    // 1 :: 2 :: 3 :: 1 :: 3 :: 1 :: 1 :: 2 

    Observable<Integer> just = Observable.fromIterable(Arrays.asList(1, 2, 3, 1, 3, 1, 1, 2)).share(); 

    Subject<Integer> objectPublishSubject = PublishSubject.<Integer>create().toSerialized(); 

    just.doOnNext(objectPublishSubject::onNext) 
      .window(() -> { 
       return objectPublishSubject.takeUntil(integer -> integer == 1).takeLast(1); 
      }).flatMap(integerObservable -> integerObservable.reduce((integer, integer2) -> { 
     return integer + integer2; 
    }).toObservable()) 
      .test().assertResult(6, 4, 1, 3); 
}