Transform RealmResult в одном фоновом потоке с RxJava

Question

Я работаю с RxJava + Realm и хотел бы сделать что-то, что кажется довольно простой:

Я хочу, чтобы загрузить все RealmObjects, которые соответствуют определенным критериям, преобразование каждого из RealmResult объекты, а затем уведомлять основной поток, когда эти преобразованные объекты готовы. Причина, по которой мне нужно преобразование в фоновом потоке, а также длительная работа, если в RealmResults имеется большое количество объектов

Проблема в том, что я могу загрузить объекты в фоновом режиме , но не может ничего сделать с каждым из объектов, все еще находясь в фоновом потоке. Я также пытаюсь сделать это в реактивной манере.

код до сих пор:

groupContactSubscription = realm.where(GroupContact.class) 
       .equalTo(MODEL_ID, mGroupId) 
       .findAllAsync() 
       .asObservable() 
       .filter(new Func1<RealmResults<GroupContact>, Boolean>() { 
        @Override 
        public Boolean call(RealmResults<GroupContact> groupContacts) { 
         return groupContacts.isLoaded(); 
        } 
       }) 
       .first() 
       .map(new Func1<RealmResults<GroupContact>, List<Contact>>() { 
        @Override 
        public List<Contact> call(RealmResults<GroupContact> groupContacts) { 
         List<Contact> contacts = new ArrayList<>(); 

         //Transform each GroupContact 
         for(int i=0; i<groupContacts.size(); ++i){ 
          contacts.add(transformGroupContact(groupContacts.get(0)) 
         } 
         return contacts; 
        } 
       }) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(new Action1<List<Contact>>() { 
        @Override 
        public void call(List<Contact> contacts) { 
         // Deal with transformed objects 
        } 
       }); 

Проблема в том, что это не происходит в фоновом потоке (она по-прежнему блокирует основной поток).

Я не совсем уверен, как это должно быть сделано чисто, может ли кто-нибудь указать мне в правильном направлении, что случилось с тем, что у меня уже есть? Я считаю, что транзакция Realm происходит асинхронно, а filter-> map также происходит на том же фоновом потоке. Ясно, что это не так.

Это моя попытка, используя RxJava поэтому, пожалуйста, будьте добры

Answer 1

 groupContactSubscription = Observable.fromCallable(() -> { 
       try(Realm realm = Realm.getDefaultInstance()) { 
        RealmRefresh.refreshRealm(realm); // from http://stackoverflow.com/a/38839808/2413303 
        RealmResults<GroupContact> groupContacts = realm.where(GroupContact.class) 
                   .equalTo(MODEL_ID, mGroupId) 
                   .findAll(); 
        List<Contact> contacts = new ArrayList<>(); 
        //Transform each GroupContact 
        for(int i = 0; i < groupContacts.size(); ++i) { 
         contacts.add(transformGroupContact(groupContacts.get(i)); 
        } 
        return contacts;       
       } 
      }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<List<Contact>>() { 
       @Override 
       public void call(List<Contact> contacts) { 
        // Deal with transformed objects 
       } 
      }); 

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

 groupContactSubscription = realm.where(GroupContacts.class).findAll().asObservable() 
     .observeOn(Schedulers.io()) 
     .switchMap(() -> 
      Observable.fromCallable(() -> { 
       try(Realm realm = Realm.getDefaultInstance()) { 
        RealmRefresh.refreshRealm(realm); // from http://stackoverflow.com/a/38839808/2413303 
        RealmResults<GroupContact> groupContacts = realm.where(GroupContact.class) 
                   .equalTo(MODEL_ID, mGroupId) 
                   .findAll(); 
        List<Contact> contacts = new ArrayList<>(); 
        //Transform each GroupContact 
        for(int i = 0; i < groupContacts.size(); ++i) { 
         contacts.add(transformGroupContact(groupContacts.get(i)); 
        } 
        return contacts;       
       } 
      }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<List<Contact>>() { 
       @Override 
       public void call(List<Contact> contacts) { 
        // Deal with transformed objects 
       } 
      }); 

---

Хотя с a proper Realm schema, вам даже не нужно будет делать такие преобразования (которые приводят к жадности оценки всей базы данных, беспересадочному крыло lazy-loading из окна), вы бы просто сделали

Subscription contacts = realm.where(Contact.class) 
             .equalTo("groupContact.modelId", mGroupId) 
             .findAllAsync() 
             .asObservable() 
             .filter(RealmResults::isLoaded) 
             .filter(RealmResults::isValid) 
             .subscribe(...); 

Или что-то в этом роде. Возможно, просто верните GroupContact как есть.