rx-java - RxJava : Window on count or timespan without dropping elements

Question

我想将我的流分成一定最大大小的批处理，如果一段时间后未达到此大小，请关闭批处理并开始新的批处理。为此，我尝试使用窗口(计数):

things.window(10)

但是，这会等到收到 10 个元素后才会发出新的 Observable 窗口。如果我使用窗口(时间跨度， 单位， 计数)运算符:

things.window(1, TimeUnit.SECONDS, 10)

我将释放 10 号之后且时间跨度完成之前出现的所有元素。

我想要一个类似的运算符，它不是等待时间跨度完成，而是在达到计数时发出一个新窗口。

things.windowXXX(timespan = 1s, count = 2) : Observable[T]

things:   ----o--o--o-----------o----o------->
timespan: [      )[    1s    ][      )[    -->
window 1: -----o-o-|
window 2:          -o---------|
window 3:                     --o-----o-| 

Answer 1

我找到了一种方法来解决此问题，即使用 buffer(timespan, timeunit, count) 而不是 window.我认为缓冲区和窗口具有相同的行为，只是发出列表和其他可观察量。但这里似乎有区别。

我的解决方案是使用缓冲区，然后将每个结果映射到可观察的:

observable.buffer(10, TimeUnit.MILLISECONDS, 2)
          .map(new Func1<List<String>, Observable<String>>() {
                public Observable<String> call(List<String> l) {
                    return Observable.from(l);
                }
           })

我做了一些测试来看看差异:https://gist.github.com/jcfandino/fd47277ada821f51a9d4

    observable.map(delayOn4).window(10, TimeUnit.MILLISECONDS, 2)
            .subscribe(new UpdateCountdowns("window"));
    assertTrue(itemsCount.await(100, TimeUnit.MILLISECONDS));
    assertTrue(batchesCount.await(100, TimeUnit.MILLISECONDS));

打印:

window - 1968827463: 1
window - 1968827463: 2
window - 1968827463: 3
window - 1267747034: 4
window - 1267747034: 5

请注意，它发出两个批处理而不是三个，并且第一个批处理包含三个元素而不是两个。

另一方面，通过使用缓冲区:

    observable.map(delayOn4).buffer(10, TimeUnit.MILLISECONDS, 2)
            .map(new Func1<List<String>, Observable<String>>() {
                public Observable<String> call(List<String> l) {
                    return Observable.from(l);
                }
            }).subscribe(new UpdateCountdowns("buffer"));
    assertTrue(itemsCount.await(100, TimeUnit.MILLISECONDS));
    assertTrue(batchesCount.await(100, TimeUnit.MILLISECONDS));

打印:

buffer - 1257525795: 1
buffer - 1257525795: 2
buffer - 1849466438: 3
buffer - 104886386: 4
buffer - 104886386: 5

三批最多有两个元素，由于“4”到达较晚，因此在下一批中结束。

我不知道这是否是预期的行为，或者其中可能存在错误。