使用基于发出的项目的条件重复/重新订阅 Observable [英] Repeat/Resubscribe to Observable with condition based on emitted item(s)

问题描述

我有 enum-values 将由 Observable 发出:

I have enum-values that will be emitted by an 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 值，有没有办法重新订阅或重复可观察对象?基本上是这样的:

Is there a way to resubscribe or to repeat the observable, if the NOT_INITIATED value is emitted? Basically something like:

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

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

I know the operators repeatWhen, which does not allow evaluation of the emitted items and retryWhen, which only acts if an error is emitted.

推荐答案

使用 flatMap - 下面不会无限循环等待期望值.向下滚动以获取支持循环的解决方案.

Use flatMap - the below does not loop infinitely waiting for the expected value. Scroll down for a solution that supports looping.

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));
}

循环直到我们得到期望值.

With looping till we get an expected value.

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;
}

这篇关于使用基于发出的项目的条件重复/重新订阅 Observable的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！