rx-java - RxJava 缓冲区直到更改

Question

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

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

在 RxJava 中我们可以使用 distinctUntilChanged()获得一个不同的项目，直到它发生变化，结果会像

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

同样，有没有办法缓冲项目直到更改？例如我期望像

这样的输出

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

Answer 1

您可以共享源序列，将 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)));
}

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