QueryPublisher: improve docs, add debug logging

Author: greenrobot-team

objectbox-java/src/main/java/io/objectbox/query/InternalAccess.java

@@ -29,4 +29,12 @@ public static <T> void nativeFindFirst(Query<T> query, long cursorHandle) {
         query.nativeFindFirst(query.handle, cursorHandle);
     }
 
+    /**
+     * See {@link QueryPublisher#LOG_STATES}.
+     */
+    @Internal
+    public static void queryPublisherLogStates() {
+        QueryPublisher.LOG_STATES = true;
+    }
+
 }

objectbox-java/src/main/java/io/objectbox/query/QueryPublisher.java

@@ -35,15 +35,26 @@
 import io.objectbox.reactive.SubscriptionBuilder;
 
 /**
- * A {@link DataPublisher} that subscribes to an ObjectClassPublisher if there is at least one observer.
- * Publishing is requested if the ObjectClassPublisher reports changes, a subscription is
- * {@link SubscriptionBuilder#observer(DataObserver) observed} or {@link Query#publish()} is called.
- * For publishing the query is re-run and the result delivered to the current observers.
- * Results are published on a single thread, one at a time, in the order publishing was requested.
+ * A {@link DataPublisher} that {@link BoxStore#subscribe(Class) subscribes to the Box} of its associated {@link Query}
+ * while there is at least one observer (see {@link #subscribe(DataObserver, Object)} and
+ * {@link #unsubscribe(DataObserver, Object)}).
+ * <p>
+ * Publishing is requested if the Box reports changes, a subscription is
+ * {@link SubscriptionBuilder#observer(DataObserver) observed} (if {@link #publishSingle(DataObserver, Object)} is
+ * called) or {@link Query#publish()} (calls {@link #publish()}) is called.
+ * <p>
+ * For publishing the query is re-run and the result data is delivered to the current observers.
+ * <p>
+ * Data is passed to observers on a single thread ({@link BoxStore#internalScheduleThread(Runnable)}), one at a time, in
+ * the order observers were added.
  */
 @Internal
 class QueryPublisher<T> implements DataPublisher<List<T>>, Runnable {
 
+    /**
+     * If enabled, logs states of the publisher runnable. Useful to debug a query subscription.
+     */
+    public static boolean LOG_STATES = false;
     private final Query<T> query;
     private final Box<T> box;
     private final Set<DataObserver<List<T>>> observers = new CopyOnWriteArraySet<>();
@@ -152,9 +163,11 @@ void stopAndAwait() {
      */
     @Override
     public void run() {
+        log("started");
         try {
             while (!publisherStopped) {
                 // Get all queued observer(s), stop processing if none.
+                log("checking for observers");
                 List<DataObserver<List<T>>> singlePublishObservers = new ArrayList<>();
                 boolean notifySubscribedObservers = false;
                 synchronized (publishQueue) {
@@ -173,10 +186,12 @@ public void run() {
                 }
 
                 // Query.
+                log("running query");
                 if (publisherStopped) break;  // Check again to avoid running the query if possible
                 List<T> result = query.find();
 
                 // Notify observer(s).
+                log("notifying observers");
                 for (DataObserver<List<T>> observer : singlePublishObservers) {
                     observer.onData(result);
                 }
@@ -189,6 +204,7 @@ public void run() {
                 }
             }
         } finally {
+            log("stopped");
             // Re-set if wrapped code throws, otherwise this publisher can no longer publish.
             publisherRunning = false;
             synchronized (this) {
@@ -206,4 +222,8 @@ public synchronized void unsubscribe(DataObserver<List<T>> observer, @Nullable O
         }
     }
 
+    private static void log(String message) {
+        if (LOG_STATES) System.out.println("QueryPublisher: " + message);
+    }
+
 }

tests/objectbox-java-test/src/test/java/io/objectbox/AbstractObjectBoxTest.java

@@ -45,6 +45,7 @@
 import io.objectbox.model.ExternalPropertyType;
 import io.objectbox.model.PropertyFlags;
 import io.objectbox.model.PropertyType;
+import io.objectbox.query.InternalAccess;
 
 
 import static org.junit.Assert.assertEquals;
@@ -92,6 +93,7 @@ static void printProcessId() {
     public void setUp() throws IOException {
         Cursor.TRACK_CREATION_STACK = true;
         Transaction.TRACK_CREATION_STACK = true;
+        InternalAccess.queryPublisherLogStates();
 
         // Note: is logged, so create before logging.
         boxStoreDir = prepareTempDir("object-store-test");