Transform RealmResult in same background thread with RxJava

Question

I'm working with RxJava + Realm and would like to do something that seems pretty straightforward:

I want to load all RealmObjects that match certain criteria, transform each of the RealmResult objects, then notify the main thread once those transformed objects are ready. The reason I need the transformation to happen in the background thread as well as it's a long running operation if there are a large number of objects in the RealmResults

The problem is, I'm able to load the objects in the background, but unable to do anything with each of the objects while still in the background thread. I'm also trying to do this in a reactive manner.

Code so far:

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
                    }
                });

The problem is, this isn't happening in the background thread (it still blocks the main thread).

I'm not really sure how this should be done cleanly, can someone point me in the right direction on what's wrong with what I have already? What I believe is happening is that the Realm transaction is happening asynchronously, and the filter->map is also happening on that same background thread. Clearly, it is not, though.

This is my attempt at using RxJava so please be kind

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
                }
            });

Although this doesn't auto-update, so I'm a bit hazy on this one, but I think this would work:

       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
                }
            });

---

Although with a proper Realm schema, you wouldn't even need to do such transformations (which result in eagerly evaluating the entire database, throwing lazy-loading out the window), you would just do

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

Or something like that. Maybe just return GroupContact as is.