RxJava 缓冲区直到更改

我有一个可观察的对象，它会发出大量数据，例如

[1,1,1,2,2,2,3,3,1,1,5,5……]

在RxJava中我们可以使用直到改变() http://reactivex.io/documentation/operators/distinct.html获得一个不同的项目，直到它发生变化，结果会像

[1,2,3,1,5,……]

同样有没有办法buffer直到更改为止的项目？例如我期望像这样的输出

[[1,1,1]、[2,2,2]、[3,3]、[1,1]、[5,5]......]

您可以分享源序列，应用distinctUntilChanged到一条路径，然后将驱动buffer使用的运算符Observable指示边界：

@Test
@SuppressWarnings("unchecked")
public void test() {
    Observable.fromArray(1,1,1,2,2,2,3,3,1,1,5,5)
    .compose(bufferUntilChanged(v -> v))
    .test()
    .assertResult(
            Arrays.asList(1, 1, 1),
            Arrays.asList(2, 2, 2),
            Arrays.asList(3, 3),
            Arrays.asList(1, 1),
            Arrays.asList(5, 5)
        );
}

static final <T, K> ObservableTransformer<T, List<T>> bufferUntilChanged(
        Function<T, K> keySelector) {
    return o -> o.publish(q -> q.buffer(q.distinctUntilChanged(keySelector).skip(1)));
}

The skip(1)是因为第一个项目通过distinctUntilChanged将触发一个新的缓冲区，使第一个缓冲区为空。