1

我需要检查一个无限的可观察（来自设备的事件）是否发出了一个特定的事件，让我们称之为“开始”，然后是另一个“完成”。然而，在这两个事件之间，可以接收到任意数量的不同事件，它们必须被忽略。其结果应该是 Completable.complete() ，当“Started”事件之后是“Finished”事件，在设置的超时之前是成功的。

对于这个问题，我有一个可行的解决方案，但是它看起来很丑而且太复杂，我认为可能有一个更优雅/更简单的解决方案。我当前的代码看起来像这样，我已经概括了我的代码，所以它更容易理解，基本上在这个例子中，我检查在 Flowable 在 10 秒超时之前发出数字“5”之后收到数字“8”。：

    Flowable<Long> events = Flowable.interval(1, TimeUnit.SECONDS, testScheduler)
            .publish().autoConnect(1);

    return events
            .filter(number -> number == 5)
            .firstElement()
            .concatMapCompletable(number -> {
                if (number == 5) {
                    return events
                            .filter(number2 -> number2 == 8)
                            .firstElement()
                            .concatMapCompletable(number2 -> {
                                if (number2 == 8) {
                                    return Completable.complete();
                                } else {
                                    return Completable.error(new Exception("Number 3 expected, got " + number2));
                                }
                            });
                } else {
                    return Completable.error(new Exception("Number 2 expected, got " + number));
                }
            })
            .timeout(10, TimeUnit.SECONDS, Completable.error(new Exception("Timeout!")));

编辑：我找到了一个更干净的版本，但是这似乎很奇怪，因为我使用 .filter 运算符然后在收到的第一个元素上完成，我将其发布在下面以供参考：

    Flowable<Long> events = Flowable.interval(1, TimeUnit.SECONDS, testScheduler)
            .publish().autoConnect(1);

    TestObserver testObserver = events
            .filter(number -> number == 5)
            .firstElement()
            .concatMapCompletable(number ->
                    events
                            .filter(number2 -> number2 == 8)
                            .firstElement()
                            .concatMapCompletable(number2 ->
                                    Completable.complete()))
            .timeout(10, TimeUnit.SECONDS, Completable.error(new Exception("Timeout!")))
            .test();

UPDATE2：我更满意的版本：

    Flowable<Long> events = Flowable.interval(1, TimeUnit.SECONDS, testScheduler)
            .publish().autoConnect(1);

    TestObserver testObserver = events
            .skipWhile(number -> number != 5)
            .firstElement()
            .flatMapCompletable(number -> Completable.fromObservable(events
                    .takeUntil(number2 -> number2 == 8)
                    .toObservable()
            ));
4

1 回答 1

0

我不确定你想要做什么，但你可以使用bufferorwindow运算符，如下所示：

Flowable.just(1, 2, 3, 4, 5)
        .buffer(2, 1)
        .filter(e -> e.size() > 1)
        .flatMapCompletable(e -> {
            int first = e.get(0);
            int second = e.get(1);
            if (first == 2) {
                if (second == 3) {
                    return Completable.complete();
                } else {
                    return Completable.error(new Exception("..."));
                }
            }

            return Completable.fromObservable(Observable.just(e));
        })

更新

Observable<Long> source = Observable.interval(1, TimeUnit.SECONDS)
        .share();

source
        .skipWhile(e -> e != 5)
        .flatMapCompletable(e -> Completable.fromObservable(source
                .takeUntil(x -> x == 8)
                .timeout(10, TimeUnit.SECONDS)))
        .subscribe();
于 2020-10-20T12:04:43.043 回答