RxJava - How to avoid this race condition?

Question

I'm trying to insert some data into the local database on a background thread with a Completeable inside the constructor of my ViewModel

public MainViewModel() extends ViewModel {

    public MainViewModel(){
        localRepository.insertValueIntoDatabase().subscribeOn(Schedulers.io())
                    .subscribe(() -> {
                        sharedPrefManager.setAnotherValue(true);
                    }, throwable -> {
                        Timber.e(throwable, "Failed to insert into DB");
                    });
    }
}

In my MainActivity I am also calling a method after the construction of the MainViewModel that will perform a query based on the inserted value in the constructor.

viewModel = new ViewModelProvider(this).get(MainViewModel.class);
viewModel.performQueryWithValue();

public MainViewModel() extends ViewModel {

    public MainViewModel(){
        localRepository.insertValueIntoDatabase().subscribeOn(Schedulers.io())
                    .subscribe(() -> {
                        sharedPrefManager.setAnotherValue(true);
                    }, throwable -> {
                        Timber.e(throwable, "Failed to insert into DB");
                    });
    }

    public void performQueryWithValue(){
        localRepository.getValueFromDatabase().flatMapSingle(value -> {
             if(value == 0){
                return remoteRepository.performQueryOne();
             }else{
                return remoteRepositoru.performQueryTwo();
             }

        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(result-> {
                   
                    
        }, err -> {
            
        });
    }
}

If the localRepository.insertValueIntoDatabase() takes 10 seconds to insert,

How do I make the performQueryWithValue() method wait for the completion of the insert before performing the query?

Answer 1

The issue here is not the calls themselves, but the nature of attempting 2 different, independent asynchronies calls against a local DB.

Let's assume that the local DB is Room.
Room allows READ access from multiple threads and has some internal rules for WRITE access.

Your use of .subscribeOn(Schedulers.io()) itself will not mean that the access calls will happen sequentially, with the second call waiting for the first to finish. That's is not how the Schedulers works.
All the Schedulers do in essence is to have actions happen on a specific thread with no guarantee of any kind of sequence.

To be able to guarantee that the WRITE happens before the READ, you have 2 options.

Options 1:
You force the sequences to happen one after another by only calling the READ after you are certain that the WRITE has completed. This can be achieved by using the Rx chain like below:

localRepository.insertValueIntoDatabase() // <- WRITE
        .subscribeOn(Schedulers.io())
        .flatMap { performQueryWithValue() } // <- READ
        .subscribe(result -> {
            sharedPrefManager.setAnotherValue(true);
            // DO something with the data pulled from the local DB
        }, throwable -> {
            Timber.e(throwable, "Failed to insert into DB");
        });

Option 2:
You perform the WRITE, but setup the READ as an Observation instead of the pull you are doing now. An Observation will only be triggered after new data exists, so you are guaranteed to only get new data when new data becomes available. Room, for example allows for this by defining the return from a Room query to be an Observable.

localRepository.insertValueIntoDatabase() // <- WRITE
        .subscribeOn(Schedulers.io())
        .subscribe(() -> {
            sharedPrefManager.setAnotherValue(true);
        }, throwable -> {
            Timber.e(throwable, "Failed to insert into DB");
        });


public void performQueryWithValue(){
    localRepository.observeValueFromDatabase() // <- OBSERVATION
        .flatMapSingle(value -> {
             if(value == 0){
                return remoteRepository.performQueryOne();
             }else{
                return remoteRepositoru.performQueryTwo();
             }

        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(result-> {
             // DO something with the data observed from the local DB   
        }, err -> {
            
        });
    }

Great guide to Option 2 here by Florina Muntenescu.