@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)); }
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; }