I text app to exercise Observable RealmResults

ObservableRealms.java

package com.mitchwongho.test.observablerealms;

import android.os.Bundle;
import android.support.annotation.NonNull;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import com.mitchwongho.test.observablerealms.realms.Person;

import java.util.concurrent.TimeUnit;

import io.realm.Realm;
import io.realm.RealmConfiguration;
import io.realm.Sort;
import rx.Observable;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;
import rx.subscriptions.CompositeSubscription;

public class MainActivity extends AppCompatActivity {

    public final static String TAG = MainActivity.class.getSimpleName();

    private CompositeSubscription subs;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        subs = new CompositeSubscription();

        Realm.init(this);

        RealmConfiguration realmConfig = new RealmConfiguration.Builder().build();
        Realm.setDefaultConfiguration(realmConfig);
    }

    @Override
    protected void onStart() {
        super.onStart();
    }

    /**
     * Dispatch onResume() to fragments.  Note that for better inter-operation
     * with older versions of the platform, at the point of this call the
     * fragments attached to the activity are <em>not</em> resumed.  This means
     * that in some cases the previous state may still be saved, not allowing
     * fragment transactions that modify the state.  To correctly interact
     * with fragments in their proper state, you should instead override
     * {@link #onResumeFragments()}.
     */
    @Override
    protected void onResume() {
        super.onResume();

        final Subscription subObservingRealm = observePersonRealm()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        person -> {
                            Log.d(TAG, String.format("onResume.onNext {threadId=%d,id=%s}", Thread.currentThread().getId(), person.id));
                        }
                        , Throwable::printStackTrace
                        , () -> {
                            Log.d(TAG, "onResume.onCompleted");
                        });
        subs.add(subObservingRealm);

        final Subscription subAsyncInserts = Observable.interval(500, TimeUnit.MILLISECONDS)
                .doOnNext(aLong -> insertOrUpdate(Person.create()))
                .subscribeOn(Schedulers.computation())
                .observeOn(Schedulers.computation())
                .subscribe(aLong -> {
                        }
                        , Throwable::printStackTrace
                        , () -> {
                        });
        subs.add(subAsyncInserts);

    }

    /**
     * Dispatch onPause() to fragments.
     */
    @Override
    protected void onPause() {
        super.onPause();
        subs.clear();
    }

    private void insertOrUpdate(@NonNull final Person person) {

        Log.d(TAG, String.format("insertOrUpdate #1:%d", Thread.currentThread().getId()));

        Observable.<Person>create(subscriber -> {
            try (final Realm realm = Realm.getDefaultInstance()) {

                realm.executeTransaction(realm1 -> {
                    Log.d(TAG, String.format("insertOrUpdate.executeTransaction #2:%d", Thread.currentThread().getId()));
                    final Person p = realm1.copyToRealm(person);
                    subscriber.onNext(p);
                });
                subscriber.onCompleted();
            } catch (Throwable t) {
                subscriber.onError(t);
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(person1 -> {
                            Log.d(TAG, String.format("insertOrUpdate.onNext #3:%d", Thread.currentThread().getId()));
                        }
                        , Throwable::printStackTrace
                        , () -> Log.d(TAG, String.format("insertOrUpdate.onCompleted #4:%d", Thread.currentThread().getId())));

    }

    private Observable<Person> observePersonRealm() {
        final Realm realm = Realm.getDefaultInstance();

        return realm.where(Person.class)
                .findAllSorted("timestamp", Sort.DESCENDING)
                .asObservable()
                .filter(persons -> persons.isLoaded() && !persons.isEmpty())
                .doOnNext(persons -> Log.d(TAG, String.format("observePersonRealm.doOnNext {size=%d}", persons.size())))
                .map(persons -> persons.first().getClone());
    }


}