根据发射的项目重复/重新订阅 Observable 条件答案

【问题标题】：Repeat/Resubscribe to Observable with condition based on emitted item(s)根据发射的项目重复/重新订阅 Observable 条件
【发布时间】：2017-01-09 20:51:24
【问题描述】：

我有 enum-values 将由 Observable 发出：

public enum BleGattOperationState {
    SUCCESSFUL,
    ERROR,
    NOT_INITIATED
}

public Observable<BleGattOperationState> readRssi() {
  return Observable.<BleGattOperationState>create(subscriber -> {
    if (someCondition()) {
        subscriber.onNext(BleGattOperationState.SUCCESSFUL);
    } else {
        subscriber.onNext(BleGattOperationState.NOT_INITIATED);
    }
  });
}

如果发出NOT_INITIATED 值，有没有办法重新订阅或重复可观察对象？基本上是这样的：

readRssi()
    .repeatIf(state -> state == NOT_INITIATED)
    .subscribe();

我知道运算符 repeatWhen，它不允许评估发出的项目和 retryWhen，它只在发出错误时起作用。

【问题讨论】：

你可以flatMap，如果NOT_INITIATED返回一个错误Observable，然后使用重试操作符。
对我来说似乎是一种解决方法。我认为只有在发生意外情况时才应该发出错误。但总比没有好。 :)
如果您认为 NOT_INITIATED 是一个错误，这不是“真正”的解决方法。这只是一个不同的语义观点
基本上我设计我的软件来实现这篇博文中的方法：blog.danlew.net/2015/12/08/error-handling-in-rxjava 他们说：“我想澄清许多 RxJava 初学者会犯的错误：onError 是一个极端事件，应该保留用于序列无法继续的时间。这意味着在处理当前项目时出现问题，因此以后无法处理任何项目。但是，在这种情况下，我可以处理抛出错误。

标签： android rx-java

【解决方案1】：

使用flatMap - 下面不会无限循环等待期望值。向下滚动查看支持循环的解决方案。

public static void main(String[] args) {
    Observable<String> o1 = Observable.just("1");
    Observable<String> o2 = Observable.just("2");

    Observable<String> o = System.currentTimeMillis() % 10 < 5 ? o1 : o2; // randomizer

    o.flatMap(s -> {
        if ("1".equals(s)) {
            return o2;
        } else {
            return Observable.just(s);
        }
    }).subscribe(next -> System.out.println(next));
}

循环直到我们得到一个期望值。

public static void main(String[] args) {
    Observable<String> o = getRandom(); // randomizer
    resolve(o).subscribe(next -> System.out.println(next));
}

static Observable<String> resolve(Observable<String> o){
    return o.flatMap(s -> {
        System.out.println("---"+s);
        if ("1".equals(s)) {
            return resolve(getRandom());
        } else {
            return Observable.just(s);
        }
    });
}


static Observable<String> getRandom(){
    Observable<String> o1 = Observable.just("1");
    Observable<String> o2 = Observable.just("2");
    long t = System.currentTimeMillis();
    System.out.println("getRandom: "+(t%10 < 8 ? 1 : 2));
    return t % 10 < 8 ? o1 : o2;
}

【讨论】：

但是这种方法不循环？还是？
感谢您的方法。最后我决定发出一个错误并用 retryWhen 来处理它，就像@dwursteisen 建议的那样。