Java 类io.realm.RealmChangeListener 实例源码

@Test
@RunTestInLooperThread
public void addListener_shouldBeCalledToReturnTheQueryResults() {
    final OsSharedRealm sharedRealm = getSharedRealmForLooper();
    populateData(sharedRealm);
    Table table = getTable(sharedRealm);

    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    looperThread.keepStrongReference(osResults);
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults osResults1) {
            assertEquals(osResults, osResults1);
            assertEquals(4, osResults1.size());
            sharedRealm.close();
            looperThread.testComplete();
        }
    });
}

@Test
public void addListener_shouldBeCalledWhenRefreshToReturnTheQueryResults() {
    final AtomicBoolean onChangeCalled = new AtomicBoolean(false);
    final OsSharedRealm sharedRealm = getSharedRealm();
    Table table = getTable(sharedRealm);

    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults osResults1) {
            assertEquals(osResults, osResults1);
            assertEquals(4, osResults1.size());
            sharedRealm.close();
            onChangeCalled.set(true);
        }
    });
    sharedRealm.refresh();
    assertTrue(onChangeCalled.get());
}

@Test
public void addListener_shouldBeCalledWhenRefreshAfterLocalCommit() {
    final CountDownLatch latch = new CountDownLatch(2);
    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    assertEquals(4, osResults.size()); // See `populateData()`
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults element) {
            if (latch.getCount() == 2) {
                // triggered by beginTransaction
                assertEquals(4, osResults.size());
            } else if (latch.getCount() == 1) {
                // triggered by refresh
                assertEquals(5, osResults.size());
            } else {
                fail();
            }
            latch.countDown();
        }
    });
    sharedRealm.beginTransaction();
    OsObject.createRow(table);
    sharedRealm.commitTransaction();
    sharedRealm.refresh();
    TestHelper.awaitOrFail(latch);
}

@Test
public void addListener_triggeredByRefresh() {
    final CountDownLatch latch = new CountDownLatch(1);
    OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    osResults.size();
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults element) {
            assertEquals(1, latch.getCount());
            latch.countDown();
        }
    });

    addRowAsync(sharedRealm);

    sharedRealm.waitForChange();
    sharedRealm.refresh();
    TestHelper.awaitOrFail(latch);
}

@Test
@RunTestInLooperThread
public void addListener_queryNotReturned() {
    final OsSharedRealm sharedRealm = getSharedRealmForLooper();
    populateData(sharedRealm);
    Table table = getTable(sharedRealm);

    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    looperThread.keepStrongReference(osResults);
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults osResults1) {
            assertEquals(osResults, osResults1);
            assertEquals(5, osResults1.size());
            sharedRealm.close();
            looperThread.testComplete();
        }
    });

    addRowAsync(sharedRealm);
}

@Test
@RunTestInLooperThread
public void addListener_queryReturned() {
    final OsSharedRealm sharedRealm = getSharedRealmForLooper();
    populateData(sharedRealm);
    Table table = getTable(sharedRealm);

    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    looperThread.keepStrongReference(osResults);
    assertEquals(4, osResults.size()); // Trigger the query to run.
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults osResults1) {
            assertEquals(osResults, osResults1);
            assertEquals(5, osResults1.size());
            sharedRealm.close();
            looperThread.testComplete();
        }
    });

    addRowAsync(sharedRealm);
}

@Test
@RunTestInLooperThread
public void collectionIterator_invalid_looperThread_byRemoteTransaction() {
    final OsSharedRealm sharedRealm = getSharedRealmForLooper();
    populateData(sharedRealm);
    Table table = getTable(sharedRealm);
    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    final TestIterator iterator = new TestIterator(osResults);
    looperThread.keepStrongReference(osResults);
    assertFalse(iterator.isDetached(sharedRealm));
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults element) {
            try {
                iterator.checkValid();
                fail();
            } catch (ConcurrentModificationException ignored) {
            }
            sharedRealm.close();
            looperThread.testComplete();
        }
    });

    addRowAsync(sharedRealm);
}

@Test
@RunTestInLooperThread
public void load() {
    final OsSharedRealm sharedRealm = getSharedRealmForLooper();
    looperThread.closeAfterTest(sharedRealm);
    populateData(sharedRealm);
    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults element) {
            assertTrue(osResults.isLoaded());
            looperThread.testComplete();
        }
    });
    assertFalse(osResults.isLoaded());
    osResults.load();
}

@Override
public void init() {
    RealmResults<TodoList> listResults = todoRepository.queryList(listId);
    list = listResults.get(0);
    view.initViews(list.getTitle());

    bindData();
    list.addChangeListener(new RealmChangeListener<RealmModel>() {
        @Override
        public void onChange(RealmModel element) {
            if (view != null) {
                bindData();
                view.notifyDataChanged(list.getTitle());
            }
        }
    });
}

@Override
public void init(long taskId) {
    RealmResults<Task> taskResults = repository.queryTask(taskId);
    if (taskResults.size() > 0) {
        task = taskResults.get(0);
    }
    task.addChangeListener(new RealmChangeListener<RealmModel>() {
        @Override
        public void onChange(RealmModel element) {
            task.getId();
            view.updateViews(task);
        }
    });
    view.initViews();
    view.updateViews(task);
}

public static Realm Realm() {
    if(realm == null){
        // Create a RealmConfiguration that saves the Realm file in the app's "files" directory.
        Realm.init(getAppContext());
        RealmConfiguration realmConfig = new RealmConfiguration.Builder()
                .deleteRealmIfMigrationNeeded()
                .build();
        Realm.setDefaultConfiguration(realmConfig);

        // Get a Realm instance for this thread
        realm = Realm.getDefaultInstance();

        realm.addChangeListener(new RealmChangeListener<Realm>() {
            @Override
            public void onChange(Realm element) {
                Log.d(Tag,"app realmOnChangeListener");
            }
        });
    }

    return realm;
}

@Override
public void subscribeDataChange(Object dataIdentity, final DataChangeListener listener) {
    if (dataIdentity instanceof RealmResults) {
        RealmResults<RealmObject> realmResults = (RealmResults<RealmObject>) dataIdentity;
        realmResults.addChangeListener(new RealmChangeListener<RealmResults<RealmObject>>() {
            @Override
            public void onChange(RealmResults<RealmObject> element) {
                if (listener != null) {
                    listener.onChange();
                }
            }
        });
    }
}

/**
 * Initialize recyclerview and adapters.
 * call adapter.notifyDataSetChanged if the realm for currency is updated
 * */
private void initializeViews(View rootView) {
    currencyRecycler = (RecyclerView) rootView.findViewById(R.id.converstions);
    currencyRecycler.setLayoutManager(new LinearLayoutManager(getActivity(), LinearLayoutManager
            .HORIZONTAL, false));

    marketAdapter=new MarketAdapter(getActivity(),rates);
    currencyRecycler.setAdapter(marketAdapter);

    rates.addChangeListener(new RealmChangeListener<RealmResults<Currency>>() {
        @Override
        public void onChange(RealmResults<Currency> currencies) {
            marketAdapter.notifyDataSetChanged();

        }
    });

}

public void openDatabase() {
    subscription = Observable.create(new Observable.OnSubscribe<Realm>() {
        @Override
        public void call(final Subscriber<? super Realm> subscriber) {
            final Realm observableRealm = Realm.getDefaultInstance();
            final RealmChangeListener<Realm> listener = realm -> {
                if(!subscriber.isUnsubscribed()) {
                    subscriber.onNext(observableRealm);
                }
            };
            observableRealm.addChangeListener(listener);
            subscriber.add(Subscriptions.create(() -> {
                observableRealm.removeChangeListener(listener);
                observableRealm.close();
            }));
            subscriber.onNext(observableRealm);
        }
    }).subscribeOn(looperScheduler.getScheduler()).unsubscribeOn(looperScheduler.getScheduler()).subscribe();
}

public SideArrayAdapter(Context context, final ArrayList<FeedWrapper> objects, ListView view) {
    super(context, 0, objects);
    this.objects = objects;
    new RealmChangeListener<Feed>() {
        @Override
        public void onChange(Feed element) {
            SideArrayAdapter.this.objects = new ArrayList<FeedWrapper>(Realm.getDefaultInstance().where(Feed.class).findAllSorted("order"));
            SideArrayAdapter.this.objects.addAll(Realm.getDefaultInstance().where(Category.class).findAllSorted("order"));
            Collections.sort(SideArrayAdapter.this.objects, new Comparator<FeedWrapper>() {
                @Override
                public int compare(FeedWrapper p1, FeedWrapper p2) {
                    return p1.getOrder() - p2.getOrder();// Ascending
                }

            });
            SideArrayAdapter.this.fitems = new ArrayList<>(objects);
        }
    };
    filter = new SubFilter();
    fitems = new ArrayList<>(objects);
    parentL = view;
}

public void openDatabase() {
    disposable = Observable.create((ObservableOnSubscribe<Realm>) emitter -> {
        final Realm observableRealm = Realm.getDefaultInstance();
        final RealmChangeListener<Realm> listener = realm -> {
            if(!emitter.isDisposed()) {
                emitter.onNext(observableRealm);
            }
        };
        observableRealm.addChangeListener(listener);
        emitter.setDisposable(Disposables.fromAction(() -> {
            observableRealm.removeChangeListener(listener);
            observableRealm.close();
        }));
        emitter.onNext(observableRealm);
    }).subscribeOn(looperScheduler.getScheduler()).unsubscribeOn(looperScheduler.getScheduler()).subscribe();
}

private Observable<List<Task>> createResults(QuerySelector<DbTask> querySelector) {
    return Observable.create((ObservableOnSubscribe<List<Task>>) emitter -> {
        Realm realm = Realm.getDefaultInstance();
        final RealmResults<DbTask> dbTasks = querySelector.createQuery(realm);
        final RealmChangeListener<RealmResults<DbTask>> realmChangeListener = element -> {
            if(element.isLoaded() && !emitter.isDisposed()) {
                List<Task> tasks = mapFrom(element);
                if(!emitter.isDisposed()) {
                    emitter.onNext(tasks);
                }
            }
        };
        emitter.setDisposable(Disposables.fromAction(() -> {
            if(dbTasks.isValid()) {
                dbTasks.removeChangeListener(realmChangeListener);
            }
            realm.close();
        }));
        dbTasks.addChangeListener(realmChangeListener);
    }).subscribeOn(looperScheduler.getScheduler()).unsubscribeOn(looperScheduler.getScheduler());
}

public void openDatabase() {
    disposable = Observable.create((ObservableOnSubscribe<Realm>) emitter -> {
        final Realm observableRealm = Realm.getDefaultInstance();
        final RealmChangeListener<Realm> listener = realm -> {
            if(!emitter.isDisposed()) {
                emitter.onNext(observableRealm);
            }
        };
        observableRealm.addChangeListener(listener);
        emitter.setDisposable(Disposables.fromAction(() -> {
            observableRealm.removeChangeListener(listener);
            observableRealm.close();
        }));
        emitter.onNext(observableRealm);
    }).subscribeOn(looperScheduler.getScheduler()).unsubscribeOn(looperScheduler.getScheduler()).subscribe();
}

private Observable<List<Task>> createResults(QuerySelector<DbTask> querySelector) {
    return Observable.create((ObservableOnSubscribe<List<Task>>) emitter -> {
        Realm realm = Realm.getDefaultInstance();
        final RealmResults<DbTask> dbTasks = querySelector.createQuery(realm);
        final RealmChangeListener<RealmResults<DbTask>> realmChangeListener = element -> {
            if(element.isLoaded() && !emitter.isDisposed()) {
                List<Task> tasks = mapFrom(element);
                if(!emitter.isDisposed()) {
                    emitter.onNext(tasks);
                }
            }
        };
        emitter.setDisposable(Disposables.fromAction(() -> {
            if(dbTasks.isValid()) {
                dbTasks.removeChangeListener(realmChangeListener);
            }
            realm.close();
        }));
        dbTasks.addChangeListener(realmChangeListener);
    }).subscribeOn(looperScheduler.getScheduler()).unsubscribeOn(looperScheduler.getScheduler());
}

private void initRealm() {
    realm = Realm.getDefaultInstance();
    realmResults = realm.where(ImageBean.class).equalTo("hasFace", true).findAllAsync();
    realmResults.addChangeListener(new RealmChangeListener<RealmResults<ImageBean>>() {
        @Override
        public void onChange(RealmResults<ImageBean> results) {
            if (results.size() > 0) {
                Log.e(TAG, "results size: " + results.size());
                mTvSearchHint.setVisibility(View.GONE);
                mImages.clear();
                mImages.addAll(results.subList(0, results.size()));
                if (mFaceAdapter != null) {
                    mFaceAdapter.notifyDataSetChanged();
                    Log.e(TAG, "getItemCount: " + mFaceAdapter.getItemCount());
                }
            } else {
                mTvSearchHint.setVisibility(View.VISIBLE);
            }
        }
    });
}

public RealmRecyclerViewAdapter(@NonNull Context context, @Nullable RealmResults<T> data, boolean autoUpdate) {
    //noinspection ConstantConditions
    if (context == null) {
        throw new IllegalArgumentException("Context can not be null");
    }

    this.context = context;
    this.adapterData = data;
    this.inflater = LayoutInflater.from(context);
    this.hasAutoUpdates = autoUpdate;

    this.listener = hasAutoUpdates ? new RealmChangeListener<RealmResults<T>>() {
        @Override
        public void onChange(RealmResults<T> results) {
            notifyChangesProcessor(results);
        }
    } : null;
}

private void setupAnime() {
    animeTitle = getIntent().getStringExtra("anime");
    this.anime = realm.where(Anime.class).equalTo(Anime.TITLE, animeTitle).findFirst();

    adapter = new EpisodeAdapter(anime, this);
    rv.setAdapter(adapter);

    setupBackground(Transformation.BLUR);
    animeListener = new RealmChangeListener<Anime>() {
        @Override
        public void onChange(Anime element) {
            if (element.coverURL != null) {
                setupBackground(Transformation.BLUR);
            }
        }
    };
    anime.addChangeListener(animeListener);

    if (anime.isStarred) {
        fabStar.setImageResource(R.drawable.ic_star_black_24dp);
    }
    if (anime.episodes.isEmpty()) {
        preloadIndicator.setVisibility(View.VISIBLE);
    }
    updateEpisodes();
}

private void setupListener() {
  realmChangeCategoryListener = new RealmChangeListener() {
    @Override public void onChange() {
      Timber.d("onChange(): category change");
      listCategory.clear();
      for (Category data : categoryRealmResults) {
        Timber.d("onChange(): " + data.getName() + " - " + data.getCreatedAt().toString());
        listCategory.add(data.getName());
      }
      spCategoryAdapter.notifyDataSetChanged();
    }
  };

  realmChangeBookListener = new RealmChangeListener() {
    @Override public void onChange() {
      bookRealmResults.sort("createdAt", Sort.DESCENDING);
      adapter.notifyDataSetChanged();
    }
  };
}

private <T extends RealmObject> void readAllAsync(@NonNull final ReadAction<T> readAction) {
    Realm realm = null;
    try {
        realm = Realm.getDefaultInstance();
        RealmQuery<T> query = readAction.getQuery(realm);
        final RealmResults<T> results = query.findAllAsync();
        results.addChangeListener(new RealmChangeListener() {
            @Override
            public void onChange() {
                results.removeChangeListener(this);
                readAction.onResults(results);
            }
        });
    } finally {
        if (realm != null) {
            realm.close();
        }
    }
}

@Override
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setupActionBar();

    realm = ((App)getApplication()).getDatabaseManager().getInstance();
    Log.d("REALM_PATH", realm.getPath());
    user = DatabaseHelper.getSessionUser(realm);

    user.addChangeListener(new RealmChangeListener() {
        @Override
        public void onChange() {
            if(!userChanged){
                userChanged = true;
            }
        }
    });

}

public static <T extends RealmModel> Flowable<RealmResults<T>> getRealmItems(Class clazz, HashMap<String, String> map) {
    return Flowable.create(new FlowableOnSubscribe<RealmResults<T>>() {
        @Override
        public void subscribe(FlowableEmitter<RealmResults<T>> emitter)
                throws Exception {
            Realm realm = Realm.getDefaultInstance();
            RealmQuery<T> query = realm.where(clazz);
            if (map != null) {
                for (Map.Entry<String, String> entry : map.entrySet()) {
                    query.equalTo(entry.getKey(), entry.getValue());
                }
            }
            RealmResults<T> results = query.findAll();

            final RealmChangeListener<RealmResults<T>> listener = _realm -> {
                if (!emitter.isCancelled()) {
                    emitter.onNext(results);
                }
            };
            emitter.setDisposable(Disposables.fromRunnable(() -> {
                results.removeChangeListener(listener);
                realm.close();
            }));
            results.addChangeListener(listener);
            emitter.onNext(results);
        }
    }, BackpressureStrategy.LATEST);
}

@Override
public boolean onOptionsItemSelected(MenuItem item) {
    //handle the click on the back arrow click
    switch (item.getItemId()) {
        case android.R.id.home:
            onBackPressed();
            return true;
        case R.id.item_add:
            mRealm.where(RealmSampleUserItem.class).findAllAsync().addChangeListener(new RealmChangeListener<RealmResults<RealmSampleUserItem>>() {
                @Override
                public void onChange(RealmResults<RealmSampleUserItem> userItems) {
                    //Remove the change listener
                    userItems.removeChangeListener(this);
                    //Store the primary key to get access from a other thread
                    final long newPrimaryKey = userItems.last().getIdentifier() + 1;
                    mRealm.executeTransactionAsync(new Realm.Transaction() {
                        @Override
                        public void execute(Realm realm) {
                            RealmSampleUserItem newUser = realm.createObject(RealmSampleUserItem.class, newPrimaryKey);
                            newUser.withName("Sample Realm Element " + newPrimaryKey);
                        }
                    });
                }
            });
            return true;
        default:
            return super.onOptionsItemSelected(item);
    }
}

@Test
@RunTestInLooperThread
@RunTestWithRemoteService(remoteService = SimpleCommitRemoteService.class, onLooperThread = true)
public void expectSimpleCommit() {
    looperThread.runAfterTest(remoteService.afterRunnable);
    remoteService.createHandler(Looper.myLooper());

    final SyncUser user = UserFactory.getInstance().createDefaultUser(Constants.AUTH_URL);
    String realmUrl = Constants.SYNC_SERVER_URL;
    final SyncConfiguration syncConfig = new SyncConfiguration.Builder(user,realmUrl)
            .directory(looperThread.getRoot())
            .build();
    final Realm realm = Realm.getInstance(syncConfig);
    final RealmResults<ProcessInfo> all = realm.where(ProcessInfo.class).findAll();
    looperThread.keepStrongReference(all);
    all.addChangeListener(new RealmChangeListener<RealmResults<ProcessInfo>>() {
        @Override
        public void onChange(RealmResults<ProcessInfo> element) {
            assertEquals(1, all.size());
            assertEquals("Background_Process1", all.get(0).getName());
            realm.close();
            user.logout();

            remoteService.triggerServiceStep(SimpleCommitRemoteService.stepB_closeRealmAndLogOut);

            looperThread.testComplete();
        }
    });

    remoteService.triggerServiceStep(SimpleCommitRemoteService.stepA_openRealmAndCreateOneObject);
}

public void onChange(T observer, @Nullable OrderedCollectionChangeSet changes) {
    if (listener instanceof OrderedRealmCollectionChangeListener) {
        //noinspection unchecked
        ((OrderedRealmCollectionChangeListener<T>) listener).onChange(observer, changes);
    } else if (listener instanceof RealmChangeListener) {
        //noinspection unchecked
        ((RealmChangeListener<T>) listener).onChange(observer);
    } else {
        throw new RuntimeException("Unsupported listener type: " + listener);
    }
}

public PendingRow(OsSharedRealm sharedRealm, TableQuery query, @Nullable SortDescriptor sortDescriptor,
                  final boolean returnCheckedRow) {
    this.sharedRealm = sharedRealm;
    pendingOsResults = OsResults.createFromQuery(sharedRealm, query, sortDescriptor, null);

    listener = new RealmChangeListener<PendingRow>() {
        @Override
        public void onChange(PendingRow pendingRow) {
            notifyFrontEnd();
        }
    };
    pendingOsResults.addListener(this, listener);
    this.returnCheckedRow = returnCheckedRow;
    sharedRealm.addPendingRow(this);
}

@Override
public <E> Flowable<RealmList<E>> from(Realm realm, final RealmList<E> list) {
    final RealmConfiguration realmConfig = realm.getConfiguration();
    return Flowable.create(new FlowableOnSubscribe<RealmList<E>>() {
        @Override
        public void subscribe(final FlowableEmitter<RealmList<E>> emitter) throws Exception {
            // Gets instance to make sure that the Realm is open for as long as the
            // Observable is subscribed to it.
            final Realm observableRealm = Realm.getInstance(realmConfig);
            listRefs.get().acquireReference(list);
            final RealmChangeListener<RealmList<E>> listener = new RealmChangeListener<RealmList<E>>() {
                @Override
                public void onChange(RealmList<E> results) {
                    if (!emitter.isCancelled()) {
                        emitter.onNext(list);
                    }
                }
            };
            list.addChangeListener(listener);

            // Cleanup when stream is disposed
            emitter.setDisposable(Disposables.fromRunnable(new Runnable() {
                @Override
                public void run() {
                    list.removeChangeListener(listener);
                    observableRealm.close();
                    listRefs.get().releaseReference(list);
                }
            }));

            // Emit current value immediately
            emitter.onNext(list);

        }
    }, BACK_PRESSURE_STRATEGY);
}

@Override
public <E> Flowable<RealmList<E>> from(DynamicRealm realm, final RealmList<E> list) {
    final RealmConfiguration realmConfig = realm.getConfiguration();
    return Flowable.create(new FlowableOnSubscribe<RealmList<E>>() {
        @Override
        public void subscribe(final FlowableEmitter<RealmList<E>> emitter) throws Exception {
            // Gets instance to make sure that the Realm is open for as long as the
            // Observable is subscribed to it.
            final DynamicRealm observableRealm = DynamicRealm.getInstance(realmConfig);
            listRefs.get().acquireReference(list);
            final RealmChangeListener<RealmList<E>> listener = new RealmChangeListener<RealmList<E>>() {
                @Override
                public void onChange(RealmList<E> results) {
                    if (!emitter.isCancelled()) {
                        emitter.onNext(list);
                    }
                }
            };
            list.addChangeListener(listener);

            // Cleanup when stream is disposed
            emitter.setDisposable(Disposables.fromRunnable(new Runnable() {
                @Override
                public void run() {
                    list.removeChangeListener(listener);
                    observableRealm.close();
                    listRefs.get().releaseReference(list);
                }
            }));

            // Emit current value immediately
            emitter.onNext(list);

        }
    }, BACK_PRESSURE_STRATEGY);
}

@Override
public <E extends RealmModel> Flowable<E> from(final Realm realm, final E object) {
    final RealmConfiguration realmConfig = realm.getConfiguration();
    return Flowable.create(new FlowableOnSubscribe<E>() {
        @Override
        public void subscribe(final FlowableEmitter<E> emitter) throws Exception {
            // Gets instance to make sure that the Realm is open for as long as the
            // Observable is subscribed to it.
            final Realm observableRealm = Realm.getInstance(realmConfig);
            objectRefs.get().acquireReference(object);
            final RealmChangeListener<E> listener = new RealmChangeListener<E>() {
                @Override
                public void onChange(E obj) {
                    if (!emitter.isCancelled()) {
                        emitter.onNext(obj);
                    }
                }
            };
            RealmObject.addChangeListener(object, listener);

            // Cleanup when stream is disposed
            emitter.setDisposable(Disposables.fromRunnable(new Runnable() {
                @Override
                public void run() {
                    RealmObject.removeChangeListener(object, listener);
                    observableRealm.close();
                    objectRefs.get().releaseReference(object);
                }
            }));

            // Emit current value immediately
            emitter.onNext(object);

        }
    }, BACK_PRESSURE_STRATEGY);
}

@Override
public Flowable<DynamicRealmObject> from(DynamicRealm realm, final DynamicRealmObject object) {
    final RealmConfiguration realmConfig = realm.getConfiguration();
    return Flowable.create(new FlowableOnSubscribe<DynamicRealmObject>() {
        @Override
        public void subscribe(final FlowableEmitter<DynamicRealmObject> emitter) throws Exception {
            // Gets instance to make sure that the Realm is open for as long as the
            // Observable is subscribed to it.
            final DynamicRealm observableRealm = DynamicRealm.getInstance(realmConfig);
            objectRefs.get().acquireReference(object);
            final RealmChangeListener<DynamicRealmObject> listener = new RealmChangeListener<DynamicRealmObject>() {
                @Override
                public void onChange(DynamicRealmObject obj) {
                    if (!emitter.isCancelled()) {
                        emitter.onNext(obj);
                    }
                }
            };
            RealmObject.addChangeListener(object, listener);

            // Cleanup when stream is disposed
            emitter.setDisposable(Disposables.fromRunnable(new Runnable() {
                @Override
                public void run() {
                    RealmObject.removeChangeListener(object, listener);
                    observableRealm.close();
                    objectRefs.get().releaseReference(object);
                }
            }));

            // Emit current value immediately
            emitter.onNext(object);

        }
    }, BACK_PRESSURE_STRATEGY);
}

@Test
@RunTestInLooperThread
public void addListener_triggeredByLocalCommit() {
    final OsSharedRealm sharedRealm = getSharedRealmForLooper();
    populateData(sharedRealm);
    Table table = getTable(sharedRealm);
    final AtomicInteger listenerCounter = new AtomicInteger(0);

    final OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    looperThread.keepStrongReference(osResults);
    osResults.addListener(osResults, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults osResults1) {
            switch (listenerCounter.getAndIncrement()) {
                case 0:
                    assertEquals(4, osResults1.size());
                    break;
                case 1:
                    assertEquals(5, osResults1.size());
                    sharedRealm.close();
                    break;
                default:
                    fail();
                    break;
            }
        }
    });
    addRow(sharedRealm);
    assertEquals(2, listenerCounter.get());
    looperThread.testComplete();
}

@Test
public void createSnapshot() {
    OsResults osResults = OsResults.createFromQuery(sharedRealm, table.where());
    OsResults snapshot = osResults.createSnapshot();
    assertTrue(OsResults.Mode.TABLEVIEW == snapshot.getMode());
    thrown.expect(IllegalStateException.class);
    snapshot.addListener(snapshot, new RealmChangeListener<OsResults>() {
        @Override
        public void onChange(OsResults element) {
        }
    });
}

@Override
public void init() {
    view.initViews();
    listsResults = repository.getAllLists();
    useFirstListAsDefault();

    listsResults.addChangeListener(new RealmChangeListener<RealmResults<TodoList>>() {
        @Override
        public void onChange(RealmResults<TodoList> element) {
            if (view != null) {
                useFirstListAsDefault();
            }
        }
    });
}

@Override
public void loadAllLists() {
    allLists = todoRepository.getAllLists();
    allLists.addChangeListener(new RealmChangeListener<RealmResults<TodoList>>() {
        @Override
        public void onChange(RealmResults<TodoList> element) {
            LogUtils.d("onChange: " + element.size());
            if (view != null) {
                view.notifyDataChanged();
            }
        }
    });

    view.bindListData(allLists);
}

public void getRealmData() {
    RealmChangeListener<RealmResults<RealmEpisode>> callback = new RealmChangeListener<RealmResults<RealmEpisode>>() {
        @Override
        public void onChange(RealmResults<RealmEpisode> element) {
            if (episodes.isLoaded()) {
                getUpcoming(episodes);
                episodes.removeChangeListeners();
            }
        }
    };

    Realm realm = RealmSingleton.getInstance().getRealm();
    Calendar c = Calendar.getInstance();
    c.setTime(new Date());
    long today = c.getTimeInMillis();

    if (DataHelper.TAB_3_INFINITE_TIME_FRAME) {

        episodes = realm.where(RealmEpisode.class)
                .greaterThan("seasonNumber", 0)
                .greaterThan("airDateTime", today)
                .findAllSortedAsync("airDateTime", Sort.ASCENDING);

    } else {

        c.add(Calendar.DATE, 30);
        long later = c.getTimeInMillis();

        episodes = realm.where(RealmEpisode.class)
                .greaterThan("seasonNumber", 0)
                .between("airDateTime", today, later)
                .findAllSortedAsync("airDateTime", Sort.ASCENDING);
    }

    episodes.addChangeListener(callback);
}

@Override
protected void onCreate(Bundle savedInstanceState) {
    Themes.applyTheme(this);
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_post);
    boot();

    post = getPost();

    post.addChangeListener(new RealmChangeListener<Post>() {
        @Override
        public void onChange(Post element) {
            Log.d(App.Tag, "post onChange => " + String.valueOf(element.isFavourite()));
        }
    });

    //set the title
    postTitle.setText(post.getTitle());

    //set the pub date
    pubDate.setText(getHumanFriendlyTime(post.getPubDate()));

    //set the Image
    Glide.with(App.getAppContext())
            .load(post.getCoverImage())
            .into(postImage);


    //load the content
    webView.loadDataWithBaseURL("file:///android_asset/", post.getDescription(), "text/html", "utf-8", null);

    //setupFavIcon();
    Log.d(App.Tag,"on create");

}