Merge pull request #97 from square/jw/slow-scheduler

Guard against a slow scheduler with backpressure strategy.

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) {