Поток сна в Schedulers.io - RxJava

Question

Я сейчас тестирую RxJava, и я запутался.
Я поместил метод с Thread.sleep (5000)

в Schedulers.io ()

CompositeDisposable compositeDisposable = new CompositeDisposable();
        compositeDisposable.add(longOperation()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableCompletableObserver() {
            @Override
            public void onComplete() {
                customToast("long operation done");
            }

            @Override
            public void onError(Throwable e) {

            }
        }));

Я прочитал, мы должны перенести длинную операцию в schedulers.io, потому что мы не можем заморозить поток пользовательского интерфейса, но в этом случае я получил пользовательский интерфейс замораживания. Что не так?
longOperation () имеет Thread.sleep (5000) внутри.

// после редактирования Callable

private void doSomething() throws InterruptedException {
        CompositeDisposable compositeDisposable = new CompositeDisposable();
        compositeDisposable.add(Observable.fromCallable(()-> longOperation())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableCompletableObserver() {
            @Override
            public void onComplete() {
                customToast("long operation done");
            }

            @Override
            public void onError(Throwable e) {

            }
        }));
    }

  private Completable longOperation() throws InterruptedException {
        Thread.sleep(5000);
        return Completable.complete();
    }

Answer 1

Вы должны использовать defer(), чтобы он действительно выполнялся в планировщике, а не когда вы пытаетесь создать завершаемый файл.

private void doSomething() throws InterruptedException {
    CompositeDisposable compositeDisposable = new CompositeDisposable();
    compositeDisposable.add(Completable.defer(new Callable<CompletableSource>() {
        @Override
        public CompletableSource call() throws Exception {
            return longOperation();
        }
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableCompletableObserver() {
                @Override
                public void onComplete() {
                    customToast("long operation done");
                }

                @Override
                public void onError(Throwable e) {

                }
            }));
}

private Completable longOperation() throws InterruptedException {
    Thread.sleep(5000);
    return Completable.complete();
}