反应式扩展：按起始对象拆分

Question

假设我有一个整数的 Observable：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) 一起使用，但它似乎有两个问题：

• 它不适用于冷的 observables
• 如果我的bufferClosingSelector 每次找到1 时都会“关闭”缓冲区，我的偏移量将为1 并发出类似1 :: 2 + 3 + 1 :: 3 + 1 :: 1 的内容，对吧？

Answer 1

我用一个主题和一个窗口解决了这个问题，但它看起来并不那么好。请看：

@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);
}