Proper usage of Schedulers

Question

I'm trying to read a file from the file system, update a room database and update the UI in the end. This seems a particularly simple task but I'm struggling with proper usage of Android and RxJava Schedulers.

@Override
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_welcome);

    this.loadingTextView = this.findViewById(R.id.tvLoading);

    this.subscription = Observable.just(0)
            .subscribeOn(Schedulers.io())
            .flatMap(ignore -> engine.checkEngineRequirements())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(ignore-> Log.e("glog", "Received on thread " + Thread.currentThread().getName()),
                    throwable -> Log.e("glog", "[WelcomeActivity.onCreate()]:" + throwable.getMessage()),
                    () -> {
                        this.loadingTextView.setText("Loaded");
                        Log.d("glog", "[WelcomeActivity] Launching new activity");
                        Intent intent = new Intent(this, PlayGameActivity.class);
                        this.startActivity(intent);
                    }
            );
}   

In checkEngineRequirements I'm loading files from a file and populating a room database:

private Observable<Long> checkEngineRequirements() {
    return new OfflineDataLoad().load(this.context)                
            .observeOn(Schedulers.io())
            .flatMap(data -> this.dataRepository.addData(data));
}

public Observable<String> load(Context context) {
    return Observable.create(emitter -> {
        try {
            BufferedReader bufferedReader = new BufferedReader(
                    new InputStreamReader(
                            context.getAssets().open(DB_FILE_NAME), "UTF-8"));
            (...)
            while (lineAvailabe) {
                emitter.onNext(line)
            }
            emitter.onComplete();
        }
        catch (Exception e) {
            emitter.onError(e);
        }
    });
}

My datasource (repository) is implemented as follows:

@Override
public Observable<Long> addData(Data newDataItem) {
    return Observable.fromCallable(() -> this.dataDao.insertData(newDataItem));
}

The problem is that the onComplete event is never called. Any idea what am I doing wrong ?

Answer 1

Based on the comments:

The general way of troubleshooting sequences is to use the doOnXXXX operators and log the various events.

In this case, the flow described did not imply any infinite sequences (as it is a very common with Room queries) and did nothing particularly complicated. Thus what remained is the stored subscription variable and the fact that the main part of the code runs in the background when started from within onCreate on the main thread and may eventually return to the main thread again.

This gives an ample time for the UI to get destroyed, and with it, the aforementioned subscription disposed. To be sure, of course, applying doOnDispose() with logging as close to the end consumer should and has revealed that indeed this is the case:

this.subscription = Observable.just(0)
        .subscribeOn(Schedulers.io())
        .flatMap(ignore -> engine.checkEngineRequirements())
        .observeOn(AndroidSchedulers.mainThread())
        // v---------------------------------v-----------------------------------v----
        .doOnDispose(() -> Log.e("glog", "[WelcomeActivity.onCreate()]: Disposed"))
        // ^---------------------------------^-----------------------------------^--
        .subscribe(ignore-> Log.e("glog", "Received on thread " 
                     + Thread.currentThread().getName()),
                throwable -> Log.e("glog", "[WelcomeActivity.onCreate()]:" 
                     + throwable.getMessage()),
                () -> {
                    this.loadingTextView.setText("Loaded");
                    Log.d("glog", "[WelcomeActivity] Launching new activity");
                    Intent intent = new Intent(this, PlayGameActivity.class);
                    this.startActivity(intent);
                }
        );