RxJava 运算符，用于将随机大小的列表流分割为统一大小的列表，而不会丢失项目

Question

我有一个项目列表流(即每个发射都是一个列表(。列表的长度各不相同。我想要一个运算符(或几个组合(可以重新分段列表，以便每个列表的长度相同。

下面是一个包含数组的简单示例。

int[] array1 = {1, 2, 3};
int[] array2 = {4, 5};
int[] array3 = {6, 7, 8, 9};
Observable.just(array1, array2, array3)
.emitGroupsOf2() // <--- what operator can I use here?
.subscribe(ints -> {
// I want:
// emit {1,2}
// emit {3,4}
// emit {5,6}
// emit {7,8}
// complete
});
}

在这个例子中，我想"坚持 3"直到我听到 4;然后我想"坚持 5"直到我听到 6。最后，我想"扔掉"9，因为我们从来没有另一个值来创建另一个对。

我看过map，scan和buffer但这些都不太正确。

Answer 1

RxJava 不支持原始数组，因此您必须执行一些包装和解包：

Observable.just(array1, array2, array3)
.concatMap(array -> Observable.range(0, array.length).map(idx -> array[idx]))
.buffer(2)
.map(buf -> {
int[] result = new int[buf.size()];
for (int i = 0; i < result.length; i++) {
result[i] = buf.get(i);
}
return result;
})
.subscribe(ints -> System.out.println(Arrays.toString(ints)));