tryingthingsout.java

package patrykpoborca.io.testapp;

import android.content.Context;
import android.os.Bundle;
import android.os.Looper;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.LinearLayout;
import android.widget.TextView;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import io.realm.Realm;
import io.realm.RealmModel;
import io.realm.RealmResults;

public class MainActivity extends AppCompatActivity {

    List<Person> people;
    private RealmWrapper realmwrapper;

    {
        people = new ArrayList<>(10);
        for (int i = 0; i < 10; i++) {
            people.add(new Person(i, "Bob = " + i));
        }
    }

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        realmwrapper = new RealmWrapper(this);
        setContentView(R.layout.activity_main);

        final LinearLayout layout = (LinearLayout) findViewById(R.id.content);
        realmwrapper.executeTransaction(people)
                .flatMap(res -> realmwrapper
                          .execute(realm -> realm.where(Person.class)
                                  .findAll())
                )
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(results -> {
                    for (int i = 0; i < results.size(); i++) {
                        TextView textView = new TextView(this);
                        Person pers = results.get(i);
                        textView.setText(pers.getName() + " Age = " + pers.getAge());
                        layout.addView(textView);
                    }
                });
    }

    public class RealmWrapper {

        private Realm realm;
        Scheduler scheduler = Schedulers.from(Executors.newSingleThreadExecutor());


        public void dis(int point) {
            Log.e("thread " + point, "Thread -> " + Thread.currentThread().getName() + " #" + Thread.currentThread().getId());
        }

        public RealmWrapper(Context context) {
            Realm.init(context);
            scheduler.scheduleDirect(() -> {
                Looper.prepare();
                Looper.loop();
                dis(1);
                realm = Realm.getDefaultInstance();
            });

            Observable.timer(50, TimeUnit.MILLISECONDS, scheduler)
                    .subscribe(new Consumer<Long>() {
                        boolean odd = true;
                        @Override
                        public void accept(Long aLong) throws Exception {
                            if(odd) {
                                Looper.getMainLooper().quit();
                            }
                            else {
                                Looper.loop();
                            }
                        }
                    });
        }

        public <T extends RealmModel>Observable<RealmResults<T>> execute(RealmLambda<T> lambda) {
            return Observable.create((ObservableOnSubscribe<RealmResults<T>>) e -> {
                RealmResults<T> value = lambda.doQuery(realm);
                e.onNext(value);

                value
                        .addChangeListener(e::onNext);
            })
            .subscribeOn(scheduler);
        }

        public <T extends RealmModel>Observable<List<T>> executeTransaction(List<T> realmModels) {
            dis(2);
            return Observable.create(new ObservableOnSubscribe<List<T>>() {
                @Override
                public void subscribe(ObservableEmitter<List<T>> e) throws Exception {
                    realm.executeTransaction(r -> {
                        e.onNext(r.copyToRealmOrUpdate(realmModels));
                        e.onComplete();
                    });
                }
            })
                    .subscribeOn(scheduler);
        }

        public <T extends RealmModel>Observable<T> executeTransaction(T realmModel) {
            return Observable.fromCallable(new Callable<T>() {
                @Override
                public T call() throws Exception {
                    return realm.copyToRealmOrUpdate(realmModel);
                }
            })
            .subscribeOn(scheduler);
        }

    }


    public interface RealmLambda<E extends RealmModel> {
        RealmResults<E> doQuery(Realm realm);
    }
}