Run RxWorker on background thread

Question

I need to run an Observable that does heavy processing on a background thread so the user doesn't feel the app freeze.

I've already tried overriding getBackgroundScheduler(), but the app keeps freezing.

@Override
protected Scheduler getBackgroundScheduler() {
    return Schedulers.computation();
}

And tried to subscribe using .subscribeOn(Schedulers.computation()), but got nothing.

I'm using WorkManager with RxJava.

This is my worker class:

public class Worker extends RxWorker {

    public Worker(@NonNull Context appContext, @NonNull WorkerParameters workerParams) {
        super(appContext, workerParams);
    }

    @NonNull
    @Override
    public Single<Result> createWork() {
        return Observables.getCounter()
                .map(currentValue -> Result.success()) // map is running on a background thread.
                .lastOrError();
    }
}

And this is the heavy work:

public class Observables {
    public static Observable<Integer> getCounter() {
        return Observable.just(1, 2, 3); // This is running on the main thread.
    }
}

Answer 1

Just create your single using Single.create(). This will make your heavy work run on the subscribing thread. It's good to override the onStopped() method to dispose the disposable.

Your rewritten class:

public class Worker extends RxWorker {

    private final CompositeDisposable compositeDisposable;

    public Worker(@NonNull Context appContext, @NonNull WorkerParameters workerParams) {
        super(appContext, workerParams);
        compositeDisposable = new CompositeDisposable();
    }

    @NonNull
    @Override
    public Single<Result> createWork() {
        return Single.create(emitter -> {
            compositeDisposable.add(Observables.getCounter().subscribe(
                    next -> emitter.onSuccess(Result.success()),
                    throwable -> emitter.onSuccess(Result.failure())
            ));
        });
    }

    @Override
    public void onStopped() {
        super.onStopped();
        compositeDisposable.dispose();
    }
}