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