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( @android.support.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); }  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; }
@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); }
@Test public void basicNotify() throws InterruptedException { String[] tables = {"a", "b"}; Set<String> tableSet = new HashSet<>(Arrays.asList(tables)); Flowable<Object> flowable = RxRoom.createFlowable(mDatabase, tables); CountingConsumer consumer = new CountingConsumer(); Disposable disposable = flowable.subscribe(consumer); assertThat(mAddedObservers.size(), CoreMatchers.is(1)); InvalidationTracker.Observer observer = mAddedObservers.get(0); assertThat(consumer.mCount, CoreMatchers.is(1)); observer.onInvalidated(tableSet); assertThat(consumer.mCount, CoreMatchers.is(2)); observer.onInvalidated(tableSet); assertThat(consumer.mCount, CoreMatchers.is(3)); disposable.dispose(); observer.onInvalidated(tableSet); assertThat(consumer.mCount, CoreMatchers.is(3)); }
@Test public void internalCallable() throws InterruptedException { final AtomicReference<String> value = new AtomicReference<>(null); String[] tables = {"a", "b"}; Set<String> tableSet = new HashSet<>(Arrays.asList(tables)); final Flowable<String> flowable = RxRoom.createFlowable(mDatabase, tables, new Callable<String>() { @Override public String call() throws Exception { return value.get(); } }); final CountingConsumer consumer = new CountingConsumer(); flowable.subscribe(consumer); InvalidationTracker.Observer observer = mAddedObservers.get(0); drain(); assertThat(consumer.mCount, CoreMatchers.is(0)); value.set("bla"); observer.onInvalidated(tableSet); drain(); assertThat(consumer.mCount, CoreMatchers.is(1)); observer.onInvalidated(tableSet); drain(); assertThat(consumer.mCount, CoreMatchers.is(2)); value.set(null); observer.onInvalidated(tableSet); drain(); assertThat(consumer.mCount, CoreMatchers.is(2)); }
@Test public void exception() throws InterruptedException { final Flowable<String> flowable = RxRoom.createFlowable(mDatabase, new String[]{"a"}, new Callable<String>() { @Override public String call() throws Exception { throw new Exception("i want exception"); } }); TestSubscriber<String> subscriber = new TestSubscriber<>(); flowable.subscribe(subscriber); drain(); assertThat(subscriber.errorCount(), CoreMatchers.is(1)); assertThat(subscriber.errors().get(0).getMessage(), CoreMatchers.is("i want exception")); }