RxJava wait for boolean(condition)

Question

There is method 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;
    }

}

How can I implement this behavior:

if isUpdating == true then wait until it will be changed to false and execute presented chain.

if isUpdating == false then just execute presented chain.

There is my solution:

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

But I think there is something wrong :)

Any better idea?

Answer 1

Not sure, did i get correctly your flow. But i guess you have some "place" where isUpdating is changed. If yes, you can just create final BehaviorSubject<Boolean> isUpdatingSubject = BehaviorSubject.<Boolean>create() and in this "place", instead of changing boolean variable, you do: isUpdatingSubject.onNext(false)

And

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

}

But be carefule: with such solution this Observable will emit each time, when isUpdatingSubject emits true, so it can be not cool sometimes. To prevent from it you can use .first() before flatMap()

PS You can read more about BehaviorSubject

PSS Sorry, i used to java7 syntax