Guard against a slow scheduler with backpressure strategy.

Schedulers apply backpressure upstream which previously behaved as an unbounded stream. When the scheduler worker couldn't keep up an MBPE was thrown. We already use the 'latest' strategy for slowing 'Query' emissions after the scheduler so we can apply the same thing before it.

This comes with a subtle change in that transactions will now synchronously run every filter Func before continuing. Previously the filters ran on the scheduler and did not block the thread performing the transaction. This means that the more queries that are listening at once, the slower every transaction's completing step will be. I don't expect this to have any meaningful impact since it's very fast and happening on a background thread already.

Author: JakeWharton

sqlbrite/src/androidTest/java/com/squareup/sqlbrite/BriteContentResolverTest.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import rx.Subscription;
+import rx.internal.util.RxRingBuffer;
 import rx.subscriptions.Subscriptions;
 
 import static com.google.common.truth.Truth.assertThat;
@@ -124,7 +125,7 @@ public void testUnsubscribeDoesNotTrigger() {
     assertThat(logs).isEmpty();
   }
 
-  public void testBackpressureSupported() {
+  public void testBackpressureSupportedWhenConsumerSlow() {
     contentResolver.insert(TABLE, values("key1", "val1"));
     o.doRequest(2);
 
@@ -170,6 +171,26 @@ public void testBackpressureSupported() {
     o.assertNoMoreEvents();
   }
 
+  public void testBackpressureSupportedWhenSchedulerSlow() {
+    subscription = db.createQuery(TABLE, null, null, null, null, false).subscribe(o);
+    o.assertCursor().isExhausted();
+
+    // Switch the scheduler to queue actions.
+    scheduler.runTasksImmediately(false);
+
+    // Shotgun twice as many insertions as the scheduler queue can handle.
+    for (int i = 0; i < RxRingBuffer.SIZE * 2; i++) {
+      contentResolver.insert(TABLE, values("key" + i, "val" + i));
+    }
+
+    scheduler.triggerActions();
+
+    // Assert we got all the events from the queue plus the one buffered from backpressure.
+    for (int i = 0; i < RxRingBuffer.SIZE + 1; i++) {
+      o.assertCursor(); // Ignore contents, just assert we got notified.
+    }
+  }
+
   public void testInitialValueAndTriggerUsesScheduler() {
     scheduler.runTasksImmediately(false);
 

sqlbrite/src/androidTest/java/com/squareup/sqlbrite/BriteDatabaseTest.java

@@ -37,6 +37,7 @@
 import rx.Observable;
 import rx.Subscription;
 import rx.functions.Action1;
+import rx.internal.util.RxRingBuffer;
 
 import static android.database.sqlite.SQLiteDatabase.CONFLICT_IGNORE;
 import static com.google.common.truth.Truth.assertThat;
@@ -744,7 +745,7 @@ public final class BriteDatabaseTest {
     o.assertNoMoreEvents();
   }
 
-  @Test public void backpressureSupported() {
+  @Test public void backpressureSupportedWhenConsumerSlow() {
     o.doRequest(2);
 
     db.createQuery(TABLE_EMPLOYEE, SELECT_EMPLOYEES).subscribe(o);
@@ -789,6 +790,30 @@ public final class BriteDatabaseTest {
     o.assertNoMoreEvents();
   }
 
+  @Test public void backpressureSupportedWhenSchedulerSlow() {
+    db.createQuery(TABLE_EMPLOYEE, SELECT_EMPLOYEES).subscribe(o);
+    o.assertCursor()
+        .hasRow("alice", "Alice Allison")
+        .hasRow("bob", "Bob Bobberson")
+        .hasRow("eve", "Eve Evenson")
+        .isExhausted();
+
+    // Switch the scheduler to queue actions.
+    scheduler.runTasksImmediately(false);
+
+    // Shotgun twice as many insertions as the scheduler queue can handle.
+    for (int i = 0; i < RxRingBuffer.SIZE * 2; i++) {
+      db.insert(TABLE_EMPLOYEE, employee("user" + i, "name" + i));
+    }
+
+    scheduler.triggerActions();
+
+    // Assert we got all the events from the queue plus the one buffered from backpressure.
+    for (int i = 0; i < RxRingBuffer.SIZE + 1; i++) {
+      o.assertCursor(); // Ignore contents, just assert we got notified.
+    }
+  }
+
   @Test public void badQueryThrows() {
     try {
       db.query("SELECT * FROM missing");

sqlbrite/src/main/java/com/squareup/sqlbrite/BriteContentResolver.java

@@ -120,9 +120,10 @@ public QueryObservable createQuery(@NonNull final Uri uri, @Nullable final Strin
       }
     };
     Observable<Query> queryObservable = Observable.create(subscribe) //
+        .onBackpressureLatest() // Guard against uncontrollable frequency of upstream emissions.
         .startWith(query) //
         .observeOn(scheduler) //
-        .onBackpressureLatest();
+        .onBackpressureLatest(); // Guard against uncontrollable frequency of scheduler executions.
     return new QueryObservable(queryObservable);
   }
 

sqlbrite/src/main/java/com/squareup/sqlbrite/BriteDatabase.java

@@ -53,8 +53,6 @@
  * the result of a query. Create using a {@link SqlBrite} instance.
  */
 public final class BriteDatabase implements Closeable {
-  private static final Set<String> INITIAL = Collections.emptySet();
-
   private final SQLiteOpenHelper helper;
   private final SqlBrite.Logger logger;
 
@@ -247,7 +245,7 @@ public QueryObservable createQuery(@NonNull final String table, @NonNull String
       @NonNull String... args) {
     Func1<Set<String>, Boolean> tableFilter = new Func1<Set<String>, Boolean>() {
       @Override public Boolean call(Set<String> triggers) {
-        return triggers == INITIAL || triggers.contains(table);
+        return triggers.contains(table);
       }
 
       @Override public String toString() {
@@ -268,9 +266,6 @@ public QueryObservable createQuery(@NonNull final Iterable<String> tables, @NonN
       @NonNull String... args) {
     Func1<Set<String>, Boolean> tableFilter = new Func1<Set<String>, Boolean>() {
       @Override public Boolean call(Set<String> triggers) {
-        if (triggers == INITIAL) {
-          return true;
-        }
         for (String table : tables) {
           if (triggers.contains(table)) {
             return true;
@@ -318,15 +313,16 @@ private QueryObservable createQuery(final Func1<Set<String>, Boolean> tableFilte
     };
 
     Observable<Query> queryObservable = triggers //
-        .startWith(INITIAL) // Immediately trigger the query for initial value.
-        .observeOn(scheduler) //
         .filter(tableFilter) // Only trigger on tables we care about.
         .map(new Func1<Set<String>, Query>() {
           @Override public Query call(Set<String> trigger) {
             return query;
           }
         }) //
-        .onBackpressureLatest() //
+        .onBackpressureLatest() // Guard against uncontrollable frequency of upstream emissions.
+        .startWith(query) //
+        .observeOn(scheduler) //
+        .onBackpressureLatest() // Guard against uncontrollable frequency of scheduler executions.
         .doOnSubscribe(new Action0() {
           @Override public void call() {
             if (transactions.get() != null) {