rxJava2 Single.Just() always executing in main thread. How to make it execute on another thread?

Question

in this code excerpt I am trying to process a bunch of data, but it can't be on the UI thread otherwise the experience is a possible ANR. I thought this was easily done with rxJava2, however, the data processing always runs on the main thread.

Data loading is triggered in the "presenter" like this:

void loadHistoricalDataFromFile(String filename){
    view.showProgressDialog();
    addDisposable(
            model.loadHistoricalDataObservable(filename)
                    .subscribeOn(rxSchedulers.runOnBackground())
                    .observeOn(rxSchedulers.mainThread())
                    .subscribe(loadedSuccessfully -> {
                        view.hideProgressDialog();
                        if (loadedSuccessfully){
                            view.showSnackBar(R.string.simulator_loaded_data_success, LENGTH_SHORT);
                        } else {
                            view.showSnackBar(R.string.simulator_loaded_data_fail, LENGTH_INDEFINITE);
                        }
                    }));
}

As you can see, I've used .subscribeOn(rxSchedulers.runOnBackground())

rxSchedulers.runOnBackground() is implemented as follows :

public class AppRxSchedulers implements RxSchedulers {


    public static Executor backgroundExecutor = Executors.newCachedThreadPool();
    public static Scheduler BACKGROUND_SCHEDULERS = Schedulers.from(backgroundExecutor);
    public static Executor internetExecutor = Executors.newCachedThreadPool();
    public static Scheduler INTERNET_SCHEDULERS = Schedulers.from(internetExecutor);
    public static Executor singleExecutor = Executors.newSingleThreadExecutor();
    public static Scheduler SINGLE_SCHEDULERS = Schedulers.from(singleExecutor);

    @Override
    public Scheduler runOnBackground() {
        return BACKGROUND_SCHEDULERS;
    }

    @Override
    public Scheduler io() {
        return Schedulers.io();
    }

    @Override
    public Scheduler compute() {
        return Schedulers.computation();
    }

    @Override
    public Scheduler mainThread() {
        return AndroidSchedulers.mainThread();
    }

    @Override
    public Scheduler internet() {
        return INTERNET_SCHEDULERS;
    }

    @Override
    public Scheduler single() {
        return SINGLE_SCHEDULERS;
    }
}

the Single.Just() is implemented as follows

Single<Boolean> loadHistoricalDataObservable(String filename){
    return Single.just(loadHistoricalData(filename));
}

private Boolean loadHistoricalData(String filename){
    boolean successful = false;
    String json = FileUtils.readFileAsStringFromExtRam(filename);
    if (json.length() > 0) {
        Gson gson = new Gson();
        historicPriceList = null;
        historicPriceList = gson.fromJson(json, new TypeToken<List<HistoricPrice>>(){}.getType());
        successful = true;
        Timber.d("Successfully loaded file - recreated %d records", historicPriceList.size());
    } else {
        Timber.d("Failed to load file");
    }

    return successful;
}

the major problem is that whenever I hit a breakpoint within loadHistoricalData() I can see it runs on the main thread. It absolutely has to be on another thread. How is this possible ?

Answer 1

The problem is here Single.just(loadHistoricalData(filename));

You are calling the function immediately and then you are passing its result to Single.just(); You need to change it to something like this:

Single.fromCallable(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return loadHistoricalData(filename);
            }
        });

So it will look like this:

Single<Boolean> loadHistoricalDataObservable(String filename){
    return Single.fromCallable(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    return loadHistoricalData(filename);
                }
            });
}

Answer 2

Single.just statement itself always runs from thread it's called from. You need use this to run your operations on different thread:

Single.create<Boolean> {
    val data = loadHistoricalData(*****)
    it.onSuccess(data)
}