RxJava 2 и работа с Realm на IO-потоке

Question

Я создаю метод внутри DataManager, который сначала загружает данные из кеша, затем запрашивает данные из API-интерфейса сервера, сохраняет результат и отправляет данные из сетевых данных ведущему (в MVP) ,

Проблема в том, что работа с Realm происходит в потоке пользовательского интерфейса, когда я хотел бы сделать его на фоне одного. Я нашел несколько статей о поддержке области RxJava, но мы используем вторую версию с другим API, поэтому эти методы Realm не помогают нам (toObservable()).

Как решить проблему?

Кроме того, как я вижу, все остальные методы обрабатываются на резьбе IO, и только Realm работает на Ui, независимо от того, что я положил subscribeOn(Schedulers.io()). Почему так происходит?

@Override 
public Observable<ChatsRepoAnswerModel> getChats() { 
    return getChatsFromCache(STATUS_OK) 
      .subscribeOn(Schedulers.io()) 
      .mergeWith(
        getChatsService() 
          .getChats() 
          .subscribeOn(Schedulers.io()) 
          .map(ChatResponseModel::getResult) 
          .flatMap(mChatsMapper::transformAll) 
          .doOnNext(this::saveChats) 
          .doOnNext(Collections::sort)        
          .onErrorResumeNext(getChatsFromCache(STATUS_ERROR))        
      .observeOn(AndroidSchedulers.mainThread()); 
} 


private void saveChats(List<ChatDataModel> realmObjects) {   
    Realm.getDefaultInstance().executeTransaction(realm -> { 
     realm.insertOrUpdate(realmObjects); 
    }); 
} 

private Observable<ChatsRepoAnswerModel> getChatsFromCache(int aStatus) { 
    Realm realm = Realm.getDefaultInstance(); 
    RealmResults<ChatDataModel> chats = realm.where(ChatDataModel.class).findAll(); 
    return processChatResponse(realm.copyFromRealm(chats), aStatus); 
} 

Answer 1

Ну а я считаю, что это полное пренебрежение к конструкции нулевой копии того, что Realm пытается дать вам:

• однонаправленного потока данных от автоматического обновления и реактивная набор данных, представленное Realm в форме RealmResults (то есть вы будете уведомлены, когда набор данных изменений)
• отложенной оценкой, только элементы переменного тока cessed в то время читаются, RealmResults просто «Курсор», и RealmObject прочитаем данные только тогда, когда аксессоров называются
• консистенция: все управляемые RealmProxies указывают на тот же объект, так что вы не «из дата»данные в любом месте (ну, кроме не-автообновления фоновых потоков, которые получают сохранить, но это вообще ошибка пользователя)

---

Это смешно, потому что realm.copyFromRealm() создает неуправляемые объекты, как правило, не имеет ни тех свойства:

• больше не Автообновления
• жадно оценить весь набор данных и копирует все данные в полях
• как результат, не обязательно до даты во всех местах, где используется

---

В любом случае решение создания отдельных RealmObjects в фоновом потоке - это открыть экземпляр этого потока, скопировать набор данных и закрыть экземпляр.

Observable.fromCallable(() -> { // <-- defer to whatever thread you're running it on 
     try(Realm realm = Realm.getDefaultInstance()) { 
      return realm.copyFromRealm(realm.where(Cat.class).findAll()); 
     } // <-- auto-close 
    }) 
.subscribeOn(Schedulers.whatever()); 

Как правило, проще использовать Realm по назначению, особенно с большими наборами данных.