Почему подписка не выполняется в новом потоке?

У меня есть такая задача:

Observable.just(getMessagesFromDb()).
    subscribeOn(Schedulers.newThread()).
    observeOn(AndroidSchedulers.mainThread()).
    subscribe(incomingMessages -> {
    //do something
    });

где getMessagesFromDb - метод, синхронно получающий сообщения, без многопоточности внутри. Согласно документации RxAndroid для метода subscribeOn:

Асинхронно подписывает Наблюдателей на это Наблюдаемое на указанном Планировщике

И есть мой вопрос: почему запрос базы данных выполняется в основном потоке? Как сделать это асинхронно?

Ответы

Ответ 1

Ниже вы видите слегка измененную версию вашего кода с вызовом getMessagesFromDb() для назначения возвращаемого значения переменной:

 Object data = getMessagesFromDb();
 Observable.just(data).
    subscribeOn(Schedulers.newThread()).
    observeOn(AndroidSchedulers.mainThread()).
    subscribe(incomingMessages -> {
      //do something
    });

Это точно, как работает Observable.just, а getMessagesFromDb() действительно вызывается в основном потоке (как вы его вызываете в основном потоке).

Вы должны отложить его с помощью метода Observable.fromCallable следующим образом:

Observable.fromCallable(() -> getMessageFromDb()).
           subscribeOn(Schedulers.newThread()).
           observeOn(AndroidSchedulers.mainThread()).
           subscribe(incomingMessages -> {
              //do something
           });

Обратите внимание, что вы можете использовать Observable.defer вместо Observable.fromCallable, но это немного сложнее:

Observable.defer(() -> Observable.just(getMessageFromDb()))

В предыдущей версии этого ответа используйте Observable.create. Но с этим методом сложно построить Observable. Старайтесь всегда предпочитать Observable.defer или Observable.fromCallable