java - RxJava等待boolean(条件)

有方法updateFromRemote()：

public class WorkshiftSettingsDaoImpl implements WorkshiftSettingsDao {

    private boolean isUpdating = false;

    public Observable<WorkshiftSettings> updateFromRemote() {
        return remoteDataStore.get()
                .retryWhen(RxOperatorsHelpers::retryWhenAnyIoExceptionWithDelay)
                .doOnSubscribe(this::setUpdatingStarted)
                .doOnUnsubscribe(this::setUpdatingFinished)
                .flatMap(workshiftSettings -> localDataStore.put(workshiftSettings));
    }

    private void setUpdatingStarted() {
        if(isUpdating) throw new RuntimeException("already updating");
        isUpdating = true;
    }

    private void setUpdatingFinished() {
        if(!isUpdating) throw new RuntimeException("already finished");
        isUpdating = false;
    }

}

如何实现此行为：

如果isUpdating == true然后等待，直到将其更改为false并执行显示的链。

如果isUpdating == false则只执行显示的链。

有我的解决方案：

public Observable<WorkshiftSettings> updateFromRemote() {
    Observable<WorkshiftSettings> updateRemoveDataObservable =  remoteDataStore.get()
            .retryWhen(RxOperatorsHelpers::retryWhenAnyIoExceptionWithDelay)
            .doOnSubscribe(this::setUpdatingStarted)
            .doOnUnsubscribe(this::setUpdatingFinished)
            .flatMap(workshiftSettings -> localDataStore.put(workshiftSettings));

    return Observable.fromCallable(() -> {
        while (isUpdating) {
            Thread.sleep(1000);
        }
        return null;
    }).concatMap(o -> updateRemoveDataObservable);
}

但我认为出了点问题:)

有更好的主意吗？

最佳答案

不确定，我是否正确获得了您的流量。但是我想您有一个isUpdating发生更改的“位置”。
如果是，则可以创建
final BehaviorSubject<Boolean> isUpdatingSubject = BehaviorSubject.<Boolean>create()
在此“位置”中，您无需更改布尔变量，而是：
 isUpdatingSubject.onNext(false)

和

public Observable<WorkshiftSettings> updateFromRemote() {
    return isUpdatingSubject
            .distinctUntilChanged()
            .filter(new Func1<Boolean, Boolean>() {
                @Override
                public Boolean call(BookingErrorActivity isUpdating) {
                    return !isUpdating;
                }
            })
            .flatMap(new Func1<Object, Observable<WorkshiftSettings>>() {
                @Override
                public Observable<WorkshiftSettings> call(Object o) {
                    return remoteDataStore.get()
                            .retryWhen(RxOperatorsHelpers::retryWhenAnyIoExceptionWithDelay)
                            .doOnSubscribe(this::setUpdatingStarted)
                            .doOnUnsubscribe(this::setUpdatingFinished)
                            .flatMap(workshiftSettings -> localDataStore.put(workshiftSettings));
                }
            });

}

但请注意：使用这种解决方案时，每次isUpdatingSubject发出true时，此Observable都会发出，因此有时可能不会很酷。
为了防止这种情况，您可以在.first()之前使用flatMap()

PS您可以阅读有关BehaviorSubject的更多信息

PSS对不起，我曾经使用过Java7语法