Observing with RxJava

The squidb-reactive module allows listening for data changes using RxJava Observables. It consists of a single class, ReactiveSquidDatabase, which users can use extend instead of SquidDatabase.

ReactiveSquidDatabase introduces six new methods:

• observeTable(SqlTable<?> table) returns an Observable that will emit the table itself each time the table is written to
• observeTableAndEmit(SqlTable<?> table, T objectToEmit) returns an Observable that will emit the given object each time the table is written to. You could for example pass a Query to be run.
• observeTablesAndEmit(Collection<SqlTable<?>> tables, T objectToEmit) returns an Observable that will emit the given object each time any of the given tables are written to
• An additional version of each of these three methods that also takes a boolean argument emitOnFirstSubscribe

Example usage:

AtomicInteger callCount = new AtomicInteger();
Observable<AtomicInteger> observable = database.observeTableAndEmit(Person.TABLE, callCount);
Subscription s = observable.subscribe(new Action1<AtomicInteger>() {
    @Override
    public void call(AtomicInteger counter) {
        counter.incrementAndGet();
    }
});
// Observable will not get notifications on first subscribe unless the observable was created using
// the emitOnInitialSubscribe flag set to true
assertEquals(0, callCount.get());
database.beginTransaction();
try {
    database.persist(new Person().setFirstName("A").setLastName("B")
                    .setBirthday(System.currentTimeMillis()));
    database.persist(new Person().setFirstName("C").setLastName("D")
                    .setBirthday(System.currentTimeMillis()));
    database.persist(new Person().setFirstName("E").setLastName("F")
                    .setBirthday(System.currentTimeMillis()));
    database.setTransactionSuccessful();
} finally {
     database.endTransaction();
}
// Count only incremented once
assertEquals(1, callCount.get());

// Unsubscribe to stop getting notifications
s.unsubscribe();

database.persist(new Person().setFirstName("E").setLastName("F")
                    .setBirthday(System.currentTimeMillis()));
// Count hasn't changed
assertEquals(1, callCount.get());

To add squidb-reactive as a dependency in build.gradle:

dependencies {
    compile 'com.yahoo.squidb:squidb-reactive:3.1.2'
}

Note that squidb-reactive includes a dependency on a particular version of RxJava, which may not match the version you would prefer to use in your own project. If you want to force a particular version of RxJava to be used in your own project, you should exclude this transitive dependency. For example:

dependencies {
    compile('com.yahoo.squidb:squidb-reactive:3.1.2') {
        exclude module: 'rxjava' // Exclude the support lib dependency by module name
    }
}

There are various other gradle strategies to exclude transitive dependencies as documented here.

---

See also:

• Listening for data changes