@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();
    // no value because it is null
    assertThat(consumer.mCount, CoreMatchers.is(0));
    value.set("bla");
    observer.onInvalidated(tableSet);
    drain();
    // get value
    assertThat(consumer.mCount, CoreMatchers.is(1));
    observer.onInvalidated(tableSet);
    drain();
    // get value
    assertThat(consumer.mCount, CoreMatchers.is(2));
    value.set(null);
    observer.onInvalidated(tableSet);
    drain();
    // no value
    assertThat(consumer.mCount, CoreMatchers.is(2));
}