Take an emission scheduler when creating wrapper types.

This has two positive effects:
- Threads which commit transactions no longer wait on downstream-triggered queries to run.
- Initial and triggered notifications now happen on the same scheduler (actually the same worker, too) which was potentially tricky to get correct before.

Author: JakeWharton

sample/src/main/java/com/example/sqlbrite/todo/db/DbModule.java

@@ -22,6 +22,7 @@
 import dagger.Module;
 import dagger.Provides;
 import javax.inject.Singleton;
+import rx.schedulers.Schedulers;
 import timber.log.Timber;
 
 @Module(complete = false, library = true)
@@ -39,7 +40,7 @@ public final class DbModule {
   }
 
   @Provides @Singleton BriteDatabase provideDatabase(SqlBrite sqlBrite, SQLiteOpenHelper helper) {
-    BriteDatabase db = sqlBrite.wrapDatabaseHelper(helper);
+    BriteDatabase db = sqlBrite.wrapDatabaseHelper(helper, Schedulers.io());
     db.setLoggingEnabled(true);
     return db;
   }

sample/src/main/java/com/example/sqlbrite/todo/ui/ItemsFragment.java

@@ -187,7 +187,6 @@ public void onViewCreated(View view, @Nullable Bundle savedInstanceState) {
             return listName + " (" + itemCount + ")";
           }
         })
-            .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
             .subscribe(new Action1<String>() {
               @Override public void call(String title) {
@@ -197,7 +196,6 @@ public void onViewCreated(View view, @Nullable Bundle savedInstanceState) {
 
     subscriptions.add(db.createQuery(TodoItem.TABLE, LIST_QUERY, listId)
         .mapToList(TodoItem.MAPPER)
-        .subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread())
         .subscribe(adapter));
   }

sample/src/main/java/com/example/sqlbrite/todo/ui/ListsFragment.java

@@ -36,7 +36,6 @@
 import javax.inject.Inject;
 import rx.Subscription;
 import rx.android.schedulers.AndroidSchedulers;
-import rx.schedulers.Schedulers;
 
 import static android.support.v4.view.MenuItemCompat.SHOW_AS_ACTION_IF_ROOM;
 import static android.support.v4.view.MenuItemCompat.SHOW_AS_ACTION_WITH_TEXT;
@@ -110,7 +109,6 @@ public void onViewCreated(View view, @Nullable Bundle savedInstanceState) {
 
     subscription = db.createQuery(ListsItem.TABLES, ListsItem.QUERY)
         .mapToList(ListsItem.MAPPER)
-        .subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread())
         .subscribe(adapter);
   }

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

@@ -40,6 +40,7 @@ public final class BriteContentResolverTest
 
   private final List<String> logs = new ArrayList<>();
   private final RecordingObserver o = new BlockingRecordingObserver();
+  private final TestScheduler scheduler = new TestScheduler();
 
   private ContentResolver contentResolver;
   private BriteContentResolver db;
@@ -59,7 +60,7 @@ public BriteContentResolverTest() {
         logs.add(message);
       }
     };
-    db = new BriteContentResolver(contentResolver, logger);
+    db = new BriteContentResolver(contentResolver, logger, scheduler);
 
     getProvider().init(getContext().getContentResolver());
   }
@@ -169,6 +170,20 @@ public void testBackpressureSupported() {
     o.assertNoMoreEvents();
   }
 
+  public void testInitialValueAndTriggerUsesScheduler() {
+    scheduler.runTasksImmediately(false);
+
+    subscription = db.createQuery(TABLE, null, null, null, null, false).subscribe(o);
+    o.assertNoMoreEvents();
+    scheduler.triggerActions();
+    o.assertCursor().isExhausted();
+
+    contentResolver.insert(TABLE, values("key1", "val1"));
+    o.assertNoMoreEvents();
+    scheduler.triggerActions();
+    o.assertCursor().hasRow("key1", "val1").isExhausted();
+  }
+
   private ContentValues values(String key, String value) {
     ContentValues result = new ContentValues();
     result.put(KEY, key);

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

@@ -58,6 +58,7 @@
 public final class BriteDatabaseTest {
   private final List<String> logs = new ArrayList<>();
   private final RecordingObserver o = new RecordingObserver();
+  private final TestScheduler scheduler = new TestScheduler();
 
   private TestDb helper;
   private SQLiteDatabase real;
@@ -72,7 +73,7 @@ public final class BriteDatabaseTest {
         logs.add(message);
       }
     };
-    db = new BriteDatabase(helper, logger);
+    db = new BriteDatabase(helper, logger, scheduler);
   }
 
   @After public void tearDown() {
@@ -162,6 +163,29 @@ public final class BriteDatabaseTest {
         .isExhausted();
   }
 
+  @Test public void queryInitialValueAndTriggerUsesScheduler() {
+    scheduler.runTasksImmediately(false);
+
+    db.createQuery(TABLE_EMPLOYEE, SELECT_EMPLOYEES).subscribe(o);
+    o.assertNoMoreEvents();
+    scheduler.triggerActions();
+    o.assertCursor()
+        .hasRow("alice", "Alice Allison")
+        .hasRow("bob", "Bob Bobberson")
+        .hasRow("eve", "Eve Evenson")
+        .isExhausted();
+
+    db.insert(TABLE_EMPLOYEE, employee("john", "John Johnson"));
+    o.assertNoMoreEvents();
+    scheduler.triggerActions();
+    o.assertCursor()
+        .hasRow("alice", "Alice Allison")
+        .hasRow("bob", "Bob Bobberson")
+        .hasRow("eve", "Eve Evenson")
+        .hasRow("john", "John Johnson")
+        .isExhausted();
+  }
+
   @Test public void queryNotNotifiedWhenInsertFails() {
     db.createQuery(TABLE_EMPLOYEE, SELECT_EMPLOYEES).subscribe(o);
     o.assertCursor()

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

@@ -24,6 +24,7 @@
 import org.junit.Test;
 import rx.functions.Func1;
 import rx.observables.BlockingObservable;
+import rx.schedulers.Schedulers;
 
 import static com.google.common.truth.Truth.assertThat;
 import static com.squareup.sqlbrite.TestDb.SELECT_EMPLOYEES;
@@ -35,7 +36,7 @@ public final class QueryTest {
   @Before public void setUp() {
     SqlBrite sqlBrite = SqlBrite.create();
     TestDb helper = new TestDb(InstrumentationRegistry.getContext());
-    db = sqlBrite.wrapDatabaseHelper(helper);
+    db = sqlBrite.wrapDatabaseHelper(helper, Schedulers.immediate());
   }
 
   @Test public void mapToOne() {

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright (C) 2016 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.squareup.sqlbrite;
+
+import java.util.concurrent.TimeUnit;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.functions.Action0;
+
+final class TestScheduler extends Scheduler {
+  private final rx.schedulers.TestScheduler delegate = new rx.schedulers.TestScheduler();
+  private boolean runTasksImmediately = true;
+
+  public void runTasksImmediately(boolean runTasksImmediately) {
+    this.runTasksImmediately = runTasksImmediately;
+  }
+
+  public void triggerActions() {
+    delegate.triggerActions();
+  }
+
+  @Override public Worker createWorker() {
+    return new TestWorker();
+  }
+
+  class TestWorker extends Worker {
+    private final Worker delegateWorker = delegate.createWorker();
+
+    @Override public Subscription schedule(Action0 action) {
+      Subscription subscription = delegateWorker.schedule(action);
+      if (runTasksImmediately) {
+        triggerActions();
+      }
+      return subscription;
+    }
+
+    @Override public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
+      Subscription subscription = delegateWorker.schedule(action, delayTime, unit);
+      if (runTasksImmediately) {
+        triggerActions();
+      }
+      return subscription;
+    }
+
+    @Override public void unsubscribe() {
+      delegateWorker.unsubscribe();
+    }
+
+    @Override public boolean isUnsubscribed() {
+      return delegateWorker.isUnsubscribed();
+    }
+  }
+}

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

@@ -27,6 +27,7 @@
 import java.util.Arrays;
 import rx.Observable;
 import rx.Observable.OnSubscribe;
+import rx.Scheduler;
 import rx.Subscriber;
 import rx.functions.Action0;
 import rx.subscriptions.Subscriptions;
@@ -45,12 +46,14 @@ public final class BriteContentResolver {
 
   final ContentResolver contentResolver;
   private final Logger logger;
+  private final Scheduler scheduler;
 
   volatile boolean logging;
 
-  BriteContentResolver(@NonNull ContentResolver contentResolver, @NonNull Logger logger) {
+  BriteContentResolver(ContentResolver contentResolver, Logger logger, Scheduler scheduler) {
     this.contentResolver = contentResolver;
     this.logger = logger;
+    this.scheduler = scheduler;
   }
 
   /** Control whether debug logging is enabled. */
@@ -67,6 +70,11 @@ public void setLoggingEnabled(boolean enabled) {
    * notifications for when the supplied {@code uri}'s data changes. Unsubscribe when you no longer
    * want updates to a query.
    * <p>
+   * Since content resolver triggers are inherently asynchronous, items emitted from the returned
+   * observable use the {@link Scheduler} supplied to {@link SqlBrite#wrapContentProvider}. For
+   * consistency, the immediate notification sent on subscribe also uses this scheduler. As such,
+   * calling {@link Observable#subscribeOn subscribeOn} on the returned observable has no effect.
+   * <p>
    * Note: To skip the immediate notification and only receive subsequent notifications when data
    * has changed call {@code skip(1)} on the returned observable.
    * <p>
@@ -113,6 +121,7 @@ public QueryObservable createQuery(@NonNull final Uri uri, @Nullable final Strin
     };
     Observable<Query> queryObservable = Observable.create(subscribe) //
         .startWith(query) //
+        .observeOn(scheduler) //
         .onBackpressureLatest();
     return new QueryObservable(queryObservable);
   }

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

@@ -33,6 +33,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import rx.Observable;
+import rx.Scheduler;
 import rx.functions.Action0;
 import rx.functions.Func1;
 import rx.subjects.PublishSubject;
@@ -52,6 +53,8 @@
  * 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;
 
@@ -99,12 +102,15 @@ public final class BriteDatabase implements Closeable {
   private volatile SQLiteDatabase writeableDatabase;
   private final Object databaseLock = new Object();
 
+  private final Scheduler scheduler;
+
   // Package-private to avoid synthetic accessor method for 'transaction' instance.
   volatile boolean logging;
 
-  BriteDatabase(@NonNull SQLiteOpenHelper helper, @NonNull SqlBrite.Logger logger) {
+  BriteDatabase(SQLiteOpenHelper helper, SqlBrite.Logger logger, Scheduler scheduler) {
     this.helper = helper;
     this.logger = logger;
+    this.scheduler = scheduler;
   }
 
   /**
@@ -223,6 +229,11 @@ public Transaction newTransaction() {
    * {@code update}, and {@code delete} methods of this class. Unsubscribe when you no longer want
    * updates to a query.
    * <p>
+   * Since database triggers are inherently asynchronous, items emitted from the returned
+   * observable use the {@link Scheduler} supplied to {@link SqlBrite#wrapDatabaseHelper}. For
+   * consistency, the immediate notification sent on subscribe also uses this scheduler. As such,
+   * calling {@link Observable#subscribeOn subscribeOn} on the returned observable has no effect.
+   * <p>
    * Note: To skip the immediate notification and only receive subsequent notifications when data
    * has changed call {@code skip(1)} on the returned observable.
    * <p>
@@ -236,7 +247,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.contains(table);
+        return triggers == INITIAL || triggers.contains(table);
       }
 
       @Override public String toString() {
@@ -257,6 +268,9 @@ 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;
@@ -304,13 +318,14 @@ 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;
           }
         }) //
-        .startWith(query) // Immediately trigger the query for initial value.
         .onBackpressureLatest() //
         .doOnSubscribe(new Action0() {
           @Override public void call() {

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

@@ -24,6 +24,7 @@
 import java.util.List;
 import rx.Observable;
 import rx.Observable.Operator;
+import rx.Scheduler;
 import rx.Subscriber;
 import rx.functions.Func1;
 
@@ -59,16 +60,24 @@ private SqlBrite(@NonNull Logger logger) {
    * interacting with the underlying {@link SQLiteOpenHelper} and it is required for automatic
    * notifications of table changes to work. See {@linkplain BriteDatabase#createQuery the
    * <code>query</code> method} for more information on that behavior.
+   *
+   * @param scheduler The {@link Scheduler} on which items from {@link BriteDatabase#createQuery}
+   * will be emitted.
    */
-  @CheckResult @NonNull
-  public BriteDatabase wrapDatabaseHelper(@NonNull SQLiteOpenHelper helper) {
-    return new BriteDatabase(helper, logger);
+  @CheckResult @NonNull public BriteDatabase wrapDatabaseHelper(@NonNull SQLiteOpenHelper helper,
+      @NonNull Scheduler scheduler) {
+    return new BriteDatabase(helper, logger, scheduler);
   }
 
-  /** Wrap a {@link ContentResolver} for observable queries. */
-  @CheckResult @NonNull
-  public BriteContentResolver wrapContentProvider(@NonNull ContentResolver contentResolver) {
-    return new BriteContentResolver(contentResolver, logger);
+  /**
+   * Wrap a {@link ContentResolver} for observable queries.
+   *
+   * @param scheduler The {@link Scheduler} on which items from
+   * {@link BriteContentResolver#createQuery} will be emitted.
+   */
+  @CheckResult @NonNull public BriteContentResolver wrapContentProvider(
+      @NonNull ContentResolver contentResolver, @NonNull Scheduler scheduler) {
+    return new BriteContentResolver(contentResolver, logger, scheduler);
   }
 
   /** An executable query. */