@Test public void basicAddRemove() { Flowable<Object> flowable = RxRoom.createFlowable(mDatabase, "a", "b"); verify(mInvalidationTracker, never()).addObserver(any(InvalidationTracker.Observer.class)); Disposable disposable = flowable.subscribe(); verify(mInvalidationTracker).addObserver(any(InvalidationTracker.Observer.class)); assertThat(mAddedObservers.size(), CoreMatchers.is(1)); InvalidationTracker.Observer observer = mAddedObservers.get(0); disposable.dispose(); verify(mInvalidationTracker).removeObserver(observer); disposable = flowable.subscribe(); verify(mInvalidationTracker, times(2)) .addObserver(any(InvalidationTracker.Observer.class)); assertThat(mAddedObservers.size(), CoreMatchers.is(2)); assertThat(mAddedObservers.get(1), CoreMatchers.not(CoreMatchers.sameInstance(observer))); InvalidationTracker.Observer observer2 = mAddedObservers.get(1); disposable.dispose(); verify(mInvalidationTracker).removeObserver(observer2); }
public static Flowable<Object> createFlowable(final RoomDatabase database, final String... tableNames) { return Flowable.create(new FlowableOnSubscribe<Object>() { @Override public void subscribe(final FlowableEmitter<Object> emitter) throws Exception { final InvalidationTracker.Observer observer = new InvalidationTracker.Observer( tableNames) { @Override public void onInvalidated(@androidx.annotation.NonNull Set<String> tables) { if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }; if (!emitter.isCancelled()) { database.getInvalidationTracker().addObserver(observer); emitter.setDisposable(Disposables.fromAction(new Action() { @Override public void run() throws Exception { database.getInvalidationTracker().removeObserver(observer); } })); } if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }, BackpressureStrategy.LATEST); }
RxRoom { public static Flowable<Object> createFlowable(final RoomDatabase database, final String... tableNames) { return Flowable.create(new FlowableOnSubscribe<Object>() { @Override public void subscribe(final FlowableEmitter<Object> emitter) throws Exception { final InvalidationTracker.Observer observer = new InvalidationTracker.Observer( tableNames) { @Override public void onInvalidated(@androidx.annotation.NonNull Set<String> tables) { if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }; if (!emitter.isCancelled()) { database.getInvalidationTracker().addObserver(observer); emitter.setDisposable(Disposables.fromAction(new Action() { @Override public void run() throws Exception { database.getInvalidationTracker().removeObserver(observer); } })); } if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }, BackpressureStrategy.LATEST); } }
RxRoom { public static Flowable<Object> createFlowable(final RoomDatabase database, final String... tableNames) { return Flowable.create(new FlowableOnSubscribe<Object>() { @Override public void subscribe(final FlowableEmitter<Object> emitter) throws Exception { final InvalidationTracker.Observer observer = new InvalidationTracker.Observer( tableNames) { @Override public void onInvalidated(@androidx.annotation.NonNull Set<String> tables) { if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }; if (!emitter.isCancelled()) { database.getInvalidationTracker().addObserver(observer); emitter.setDisposable(Disposables.fromAction(new Action() { @Override public void run() throws Exception { database.getInvalidationTracker().removeObserver(observer); } })); } if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }, BackpressureStrategy.LATEST); } @Deprecated @SuppressWarnings("PrivateConstructorForUtilityClass")  RxRoom(); }
RxRoom { public static Flowable<Object> createFlowable(final RoomDatabase database, final String... tableNames) { return Flowable.create(new FlowableOnSubscribe<Object>() { @Override public void subscribe(final FlowableEmitter<Object> emitter) throws Exception { final InvalidationTracker.Observer observer = new InvalidationTracker.Observer( tableNames) { @Override public void onInvalidated(@androidx.annotation.NonNull Set<String> tables) { if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }; if (!emitter.isCancelled()) { database.getInvalidationTracker().addObserver(observer); emitter.setDisposable(Disposables.fromAction(new Action() { @Override public void run() throws Exception { database.getInvalidationTracker().removeObserver(observer); } })); } if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }, BackpressureStrategy.LATEST); } @Deprecated @SuppressWarnings("PrivateConstructorForUtilityClass")  RxRoom(); static Flowable<Object> createFlowable(final RoomDatabase database,
            final String... tableNames); @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP) static Flowable<T> createFlowable(final RoomDatabase database,
            final String[] tableNames, final Callable<T> callable); }
RxRoom { public static Flowable<Object> createFlowable(final RoomDatabase database, final String... tableNames) { return Flowable.create(new FlowableOnSubscribe<Object>() { @Override public void subscribe(final FlowableEmitter<Object> emitter) throws Exception { final InvalidationTracker.Observer observer = new InvalidationTracker.Observer( tableNames) { @Override public void onInvalidated(@androidx.annotation.NonNull Set<String> tables) { if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }; if (!emitter.isCancelled()) { database.getInvalidationTracker().addObserver(observer); emitter.setDisposable(Disposables.fromAction(new Action() { @Override public void run() throws Exception { database.getInvalidationTracker().removeObserver(observer); } })); } if (!emitter.isCancelled()) { emitter.onNext(NOTHING); } } }, BackpressureStrategy.LATEST); } @Deprecated @SuppressWarnings("PrivateConstructorForUtilityClass")  RxRoom(); static Flowable<Object> createFlowable(final RoomDatabase database,
            final String... tableNames); @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP) static Flowable<T> createFlowable(final RoomDatabase database,
            final String[] tableNames, final Callable<T> callable); static final Object NOTHING; }