带有反跳功能的RxJava2缓冲区

Question

我想缓冲元素并在x时间内没有新元素时将它们作为集合发出。怎么做？

例如给定的输入

INPUT    TIME
1        0  
2        0
3        100
4        150
5        400
6        450
7        800   

如果我的x = 200我想发射 {1, 2, 3, 4}, {5, 6}, {7}

我尝试过的方法很简单buffer()，但是随着时间的推移却无法消除抖动。我也试过throttleFirst()源和flatMap()它buffer().take(1)的来源里面flatMap，它的工作原理类似，但不完全是任意的。

Answer 1

You need publish as you need the same source to control the buffering behavior through a debounce:

static <T> ObservableTransformer<T, List<T>> bufferDebounce(
        long time, TimeUnit unit, Scheduler scheduler) {
    return o ->
        o.publish(v -> 
            v.buffer(v.debounce(time, unit, scheduler)
                .takeUntil(v.ignoreElements().toObservable())
            )
        );
}

@Test
public void test() {
    PublishSubject<Integer> ps = PublishSubject.create();

    TestScheduler sch = new TestScheduler();

    ps.compose(bufferDebounce(200, TimeUnit.MILLISECONDS, sch))
    .subscribe(
            v -> System.out.println(sch.now(TimeUnit.MILLISECONDS)+ ": " + v),
            Throwable::printStackTrace,
            () -> System.out.println("Done"));

    ps.onNext(1);
    ps.onNext(2);

    sch.advanceTimeTo(100, TimeUnit.MILLISECONDS);

    ps.onNext(3);

    sch.advanceTimeTo(150, TimeUnit.MILLISECONDS);

    ps.onNext(4);

    sch.advanceTimeTo(400, TimeUnit.MILLISECONDS);

    ps.onNext(5);

    sch.advanceTimeTo(450, TimeUnit.MILLISECONDS);

    ps.onNext(6);

    sch.advanceTimeTo(800, TimeUnit.MILLISECONDS);

    ps.onNext(7);
    ps.onComplete();

    sch.advanceTimeTo(850, TimeUnit.MILLISECONDS);
}

The takeUntil is there to prevent the completion of o to trigger an empty buffer.