rx-java - RxJava3 : debounce/skip unwanted '0' in a flow of integers but want '0' only if getting continuously

Question

我有一个 Integer 类型的 Observable，它发出 0 和正数，还有一个 Observer 来捕捉两者。

Observable 每 10 毫秒(一次一个)发出连续的零或连续的正数。

赞 0,0,0,0,0,0,0,5,5,6,7,8,8,9,10,11,4,5,6,5,0,0,0,0,0... [预期的]

此外，它在正数之间发出零(很少)。

赞 0,0,0,0,0,0,0,5,5,6,7,8,8,**0**,9,10,11,4,5,6,5,0,0,0,0,0... [意外]

我想消除/跳过两个正数之间的这个零，但如果它是连续的，则对捕捉 0 感兴趣。

在 rxjava 中是否有任何运算符组合可以实现这一点。提前致谢。

代码如下所示:

public Observer<Integer> valueObserver = new DisposingObserver<Integer>() {
        @Override
        public void onNext(Integer value) {
         //every 10 seconds a value is received
         //do action based on zero or non-zero values.
        }
    };

Observable<Integer> sourceObservable = Observable.just(0,0,0,0,0,0,0,5,5,6,7,8,8,0,9,10,11,4,5,6,5,0,0,0,0,0,...);
sourceObservable.subscribe(valueObserver);

预期o/p:0,5,6,7,8,9,10,11,4,5,6,5,0

我可以通过使用 distinctUntilChanged 运算符来消除连续的重复元素。

Answer 1

这需要记住到目前为止流已经产生了多少个后续零并且什么都不发出，一个零、两个零或非零项。你可以用 flatMapIterable 来做到这一点但是计数零是有状态的，因此需要 defer :

static ObservableTransformer<Integer, Integer> skipSingleZero() {
    return source -> Observable.defer(() -> {
        AtomicInteger zerosSeen = new AtomicInteger();

        return source
        .flatMapIterable(item -> {
            if (item == 0) {
                int n = zerosSeen.getAndIncrement();
                // first zero we saw, don't emit it
                if (n == 0) {
                    return Collections.emptyList();
                }
                // the second zero in a row, it means a streak so emit both
                if (n == 1) {
                    return Arrays.asList(0, 0);
                }
                // third or more zeros, just emit those now on
                return Collections.singletonList(0);
            }
            // a non-zero item, reset the counter
            zerosSeen.set(0);
            return Collections.singletonList(item);
        });
    });
}

这是一个中间有一个单独零的序列，因此应该跳过:

Observable<Integer> source = Observable.fromArray(
        0, 0, 0, 0, 0, 0, 0, 5, 5, 6, 7, 8, 8,
        0, 9, 10, 11, 4, 5, 6, 5, 0, 0, 0, 0, 0
);

source
.compose(skipSingleZero())
.test()
.assertResult(
        0, 0, 0, 0, 0, 0, 0, 5, 5, 6, 7, 8, 8,
        9, 10, 11, 4, 5, 6, 5, 0, 0, 0, 0, 0
);

这中间有 2 连胜，应该保留:

Observable<Integer> source = Observable.fromArray(
        0, 0, 0, 0, 0, 0, 0, 5, 5, 6, 7, 8, 8,
        0, 0, 9, 10, 11, 4, 5, 6, 5, 0, 0, 0, 0, 0
);

source
.compose(skipSingleZero())
.test()
.assertResult(
        0, 0, 0, 0, 0, 0, 0, 5, 5, 6, 7, 8, 8,
        0, 0, 9, 10, 11, 4, 5, 6, 5, 0, 0, 0, 0, 0
);