<commit_before>package mansonheart.com.database;

import org.json.JSONArray;
import org.json.JSONObject;

import javax.inject.Provider;

import io.realm.Realm;
import io.realm.RealmObject;
import io.realm.RealmQuery;
import io.realm.RealmResults;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Func1;
import rx.subjects.BehaviorSubject;

/**
 * Created by <NAME> on 07.01.2016.
 */
public class RealmRepositoryImpl implements RealmRepository {

    Provider<Realm> realmProvider;
    RealmQueryableCollection realmQueryCollection;

    public RealmRepositoryImpl(Provider<Realm> realmProvider) {
        this.realmProvider = realmProvider;
        this.realmQueryCollection = new RealmQueryableCollection();
    }

    @Override
    public <T> Observable<T> get(Class clazz, Func1<RealmQuery, RealmQuery> predicate) {
        BehaviorSubject<T> behaviorSubject = BehaviorSubject.create((T) getInner(clazz, predicate));
        realmQueryCollection.add(clazz, predicate, behaviorSubject);
        return behaviorSubject;
    }

    public <T extends RealmObject> RealmResults<T> getInner(Class clazz, Func1<RealmQuery, RealmQuery> predicate) {
        RealmQuery query = getRealm().where(clazz);
        if (predicate != null)
            query = predicate.call(query);
        return query.findAllAsync();
    }

    @Override
    public void storeObject(Class clazz, JSONObject jsonObject) {
        getRealm().beginTransaction();
        getRealm().createOrUpdateObjectFromJson(clazz, jsonObject);
        getRealm().commitTransaction();
        notifyObservers(clazz);

    }

    @Override
    public void storeObjects(Class clazz, JSONArray jsonArray) {
        getRealm().beginTransaction();
        getRealm().createOrUpdateAllFromJson(clazz, jsonArray);
        getRealm().commitTransaction();
        notifyObservers(clazz);
    }


    @Override
    public <T> Observable<T> update(Class clazz, Action0 action) {
        return (Observable<T>) Observable.create(subscriber -> {
            getRealm().beginTransaction();
            action.call();
        }).doOnNext(o -> {
            getRealm().commitTransaction();
            notifyObservers(clazz);
        });
    }

    private void notifyObservers(Class clazz) {
        Observable.from(realmQueryCollection.getQuerables(clazz))
                .subscribe(realmQuerable ->
                        realmQuerable.getSubject().onNext(getInner(clazz, realmQuerable.getPredicate())));
    }

    private Realm getRealm() {
        return realmProvider.get();
    }
}<commit_msg>Check subject observers<commit_after>package mansonheart.com.database;

import org.json.JSONArray;
import org.json.JSONObject;

import javax.inject.Provider;

import io.realm.Realm;
import io.realm.RealmObject;
import io.realm.RealmQuery;
import io.realm.RealmResults;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Func1;
import rx.subjects.BehaviorSubject;

/**
 * Created by <NAME> on 07.01.2016.
 */
public class RealmRepositoryImpl implements RealmRepository {

    Provider<Realm> realmProvider;
    RealmQueryableCollection realmQueryCollection;

    public RealmRepositoryImpl(Provider<Realm> realmProvider) {
        this.realmProvider = realmProvider;
        this.realmQueryCollection = new RealmQueryableCollection();
    }

    @Override
    public <T> Observable<T> get(Class clazz, Func1<RealmQuery, RealmQuery> predicate) {
        BehaviorSubject<T> behaviorSubject = BehaviorSubject.create((T) getInner(clazz, predicate));
        realmQueryCollection.add(clazz, predicate, behaviorSubject);
        return behaviorSubject;
    }

    public <T extends RealmObject> RealmResults<T> getInner(Class clazz, Func1<RealmQuery, RealmQuery> predicate) {
        RealmQuery query = getRealm().where(clazz);
        if (predicate != null)
            query = predicate.call(query);
        return query.findAllAsync();
    }

    @Override
    public void storeObject(Class clazz, JSONObject jsonObject) {
        getRealm().beginTransaction();
        getRealm().createOrUpdateObjectFromJson(clazz, jsonObject);
        getRealm().commitTransaction();
        notifyObservers(clazz);
    }

    @Override
    public void storeObjects(Class clazz, JSONArray jsonArray) {
        getRealm().beginTransaction();
        getRealm().createOrUpdateAllFromJson(clazz, jsonArray);
        getRealm().commitTransaction();
        notifyObservers(clazz);
    }


    @Override
    public <T> Observable<T> update(Class clazz, Action0 action) {
        return (Observable<T>) Observable.create(subscriber -> {
            getRealm().beginTransaction();
            action.call();
        }).doOnNext(o -> {
            getRealm().commitTransaction();
            notifyObservers(clazz);
        });
    }

    private void notifyObservers(Class clazz) {
        Observable.from(realmQueryCollection.getQuerables(clazz))
                .subscribe(realmQuerable -> {
                    if (!realmQuerable.getSubject().hasObservers()) {
                        realmQueryCollection.queryables.remove(realmQuerable);
                    } else {
                        RealmResults realmResults = getInner(clazz, realmQuerable.getPredicate());
                        realmResults.load();
                        realmQuerable.getSubject().onNext(realmResults);
                    }
                });
    }

    private Realm getRealm() {
        return realmProvider.get();
    }
}
{'fraction_non_alphanumeric': 0.10469577928010232, 'fraction_numerical': 0.004750593824228029, 'mean_word_length': 3.0014619883040936, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4045161', 'n_tokens_mistral': 1573, 'n_tokens_neox': 1527, 'n_words': 286}