Merge pull request #85 from square/jw/check-on-subscribe

Check for enclosing transaction on actual subscribe, not initial emission.

Author: Alec Strong

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

@@ -546,6 +546,16 @@ public final class BriteDatabaseTest {
     o.assertErrorContains("Cannot subscribe to observable query in a transaction.");
   }
 
+  @Test public void querySubscribedToDuringTransactionThrowsWithBackpressure() {
+    o.doRequest(0);
+
+    Observable<Query> query = db.createQuery(TABLE_EMPLOYEE, SELECT_EMPLOYEES);
+
+    db.newTransaction();
+    query.subscribe(o);
+    o.assertErrorContains("Cannot subscribe to observable query in a transaction.");
+  }
+
   @Test public void callingEndMultipleTimesThrows() {
     Transaction transaction = db.newTransaction();
     transaction.end();

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.functions.Action0;
 import rx.functions.Func1;
 import rx.subjects.PublishSubject;
 
@@ -297,18 +298,22 @@ private QueryObservable createQuery(final Func1<Set<String>, Boolean> tableFilte
         .startWith(INITIAL_TRIGGER) // Immediately execute the query for initial value.
         .map(new Func1<Set<String>, Query>() {
           @Override public Query call(Set<String> trigger) {
-            if (transactions.get() != null) {
-              throw new IllegalStateException(
-                  "Cannot subscribe to observable query in a transaction.");
-            }
             if (logging) {
               log("QUERY\n  trigger: %s\n  tables: %s\n  sql: %s\n  args: %s", trigger, tableFilter,
                   sql, Arrays.toString(args));
             }
             return query;
           }
         }) //
-        .onBackpressureLatest();
+        .onBackpressureLatest() //
+        .doOnSubscribe(new Action0() {
+          @Override public void call() {
+            if (transactions.get() != null) {
+              throw new IllegalStateException(
+                  "Cannot subscribe to observable query in a transaction.");
+            }
+          }
+        });
     return new QueryObservable(queryObservable);
   }