rxJava 等待特定值序列的更好方法

Question

我想知道过滤来自可观察管道的特定值序列的正确方法是什么。例如：

我遇到了一个问题，我需要在管道变为 false 之后做一些事情，但只能在 pipe 变为 true 之后直接做。

Observable x = Observable.just(false, false, false, true, false);

我的第一个方法：

x
.filter(state -> state)
.switchMap(x)
.filter(state -> !state)
.subscribe(x -> doSomething())

第二种方法：

private boolean globalState = false;


private void someFunction() {
   x.map(this::hasTurnedOffAgain)
   .filter(state -> state)
   .subscribe(doSomething)
}

...

private boolean hasTurnedOggAgain(Boolean state) {
    boolean hasTurnedOff= !state && peeping;
    globalState = state;
    return hasFinished;
}

我有一个使用扫描运算符的第三个版本，但我认为这个简单的问题太复杂了。

有没有更好的方法来实现这样的东西？甚至更复杂的序列？

谢谢你，祝你有美好的一天！

Answer 1

当您需要 class 来保存状态而不仅仅是函数时就是这种情况。

class FilterClass {
    Boolean prev = null;

    boolean filter(Boolean current) {
        // detect false but only directly after being true.
        boolean res = (current==Boolean.FALSE) && (prev==Boolean.TRUE);
        prev = current;
        return res;
    }
}

Observable<Boolean> x = Observable.just(false, false, false, true, false);
FilterClass filter=new FilterClass();

x.filter(filter::filter)
 .subscribe(x -> doSomething(x));

Answer 2

您可以使用 buffer() 方法构建一个 Observable，该 Observable 具有原始 Observable 的上一个和当前项目。如果要检查两个相邻值，请使用 count 值 2，如果要向前迈出一步，请使用 skip 值 1。

Observable<Boolean> original = Observable.just(false, false, false, true, false, false, true, true, false);

Observable<List<Boolean>> buffered = original.buffer(2, 1);

buffered.subscribe(new Consumer<List<Boolean>>() {

    @Override
    public void accept(List<Boolean> t) throws Exception {
        System.out.println("Direct:"+t);
    }
});

这将生成以下输出：

Direct:[false, false]
Direct:[false, false]
Direct:[false, true]
Direct:[true, false]
Direct:[false, false]
Direct:[false, true]
Direct:[true, true]
Direct:[true, false]
Direct:[false]

您可以使用这个新的 observable 并使用 filter() 仅获取这些值，其中先前的条目是 true 而当前的条目是 false，因此您会看到 "falling edge":

Observable<List<Boolean>> filtered = buffered.filter(l -> l.size() == 2 &&
                                                          l.get(0) == true && 
                                                          l.get(1) == false);

filtered.subscribe(new Consumer<List<Boolean>>() {

    @Override
    public void accept(List<Boolean> t) throws Exception {
        System.out.println("Filtered: "+t);
    }
});

这将生成以下输出：

Filtered: [true, false]
Filtered: [true, false]

rxJava 等待特定值序列的更好方法

rxJava better way to wait on a specific sequence of values

java

rx-java

reactive