Query/QueryPublisher: wait for the query to finish on close #181

Otherwise, Query would destroy its native counterpart while it's still
in use.

Author: greenrobot

CHANGELOG.md

@@ -7,6 +7,7 @@ For more insights into what changed in the ObjectBox C++ core, [check the Object
 ## 4.3.1 - in development
 
 - Requires at least Kotlin compiler and standard library 1.7.
+- Data Observers: closing a Query now waits on a running publisher to finish its query, preventing a VM crash. [#1147](https://github.com/objectbox/objectbox-java/issues/1147)
 
 ## 4.3.0 - 2025-05-13
 

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

@@ -175,6 +175,7 @@ protected void finalize() throws Throwable {
      * Calling any other methods of this afterwards will throw an {@link IllegalStateException}.
      */
     public synchronized void close() {
+        publisher.stopAndAwait();  // Ensure it is done so that the query is not used anymore
         if (handle != 0) {
             // Closeable recommendation: mark as "closed" before nativeDestroy could throw.
             long handleCopy = handle;
@@ -939,6 +940,7 @@ public long remove() {
      * See {@link SubscriptionBuilder#observer(DataObserver)} for additional details.
      *
      * @return A {@link SubscriptionBuilder} to build a subscription.
+     * @see #publish()
      */
     public SubscriptionBuilder<List<T>> subscribe() {
         checkOpen();
@@ -958,11 +960,15 @@ public SubscriptionBuilder<List<T>> subscribe(DataSubscriptionList dataSubscript
     }
 
     /**
-     * Publishes the current data to all subscribed @{@link DataObserver}s.
-     * This is useful triggering observers when new parameters have been set.
-     * Note, that setParameter methods will NOT be propagated to observers.
+     * Manually schedules publishing the current results of this query to all {@link #subscribe() subscribed}
+     * {@link DataObserver observers}, even if the underlying Boxes have not changed.
+     * <p>
+     * This is useful to publish new results after changing parameters of this query which would otherwise not trigger
+     * publishing of new results.
      */
     public void publish() {
+        // Do open check to not silently fail (publisher runnable would just not get scheduled if query is closed)
+        checkOpen();
         publisher.publish();
     }
 

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

@@ -49,6 +49,7 @@ class QueryPublisher<T> implements DataPublisher<List<T>>, Runnable {
     private final Set<DataObserver<List<T>>> observers = new CopyOnWriteArraySet<>();
     private final Deque<DataObserver<List<T>>> publishQueue = new ArrayDeque<>();
     private volatile boolean publisherRunning = false;
+    private volatile boolean publisherStopped = false;
 
     private static class SubscribedObservers<T> implements DataObserver<List<T>> {
         @Override
@@ -106,6 +107,10 @@ void publish() {
      */
     private void queueObserverAndScheduleRun(DataObserver<List<T>> observer) {
         synchronized (publishQueue) {
+            // Check after obtaining the lock as the publisher may have been stopped while waiting on the lock
+            if (publisherStopped) {
+                return;
+            }
             publishQueue.add(observer);
             if (!publisherRunning) {
                 publisherRunning = true;
@@ -114,6 +119,31 @@ private void queueObserverAndScheduleRun(DataObserver<List<T>> observer) {
         }
     }
 
+    /**
+     * Marks this publisher as stopped and if it is currently running waits on it to complete.
+     * <p>
+     * After calling this, this publisher will no longer run, even if observers subscribe or publishing is requested.
+     */
+    void stopAndAwait() {
+        publisherStopped = true;
+        // Doing wait/notify waiting here; could also use the Future from BoxStore.internalScheduleThread() instead.
+        // The latter would require another member though, which seems redundant.
+        synchronized (this) {
+            while (publisherRunning) {
+                try {
+                    this.wait();
+                } catch (InterruptedException e) {
+                    if (publisherRunning) {
+                        // When called by Query.close() throwing here will leak the query. But not throwing would allow
+                        // close() to proceed in destroying the native query while it may still be active (run() of this
+                        // is at the query.find() call), which would trigger a VM crash.
+                        throw new RuntimeException("Interrupted while waiting for publisher to finish", e);
+                    }
+                }
+            }
+        }
+    }
+
     /**
      * Processes publish requests for this query on a single thread to prevent
      * older query results getting delivered after newer query results.
@@ -123,7 +153,7 @@ private void queueObserverAndScheduleRun(DataObserver<List<T>> observer) {
     @Override
     public void run() {
         try {
-            while (true) {
+            while (!publisherStopped) {
                 // Get all queued observer(s), stop processing if none.
                 List<DataObserver<List<T>>> singlePublishObservers = new ArrayList<>();
                 boolean notifySubscribedObservers = false;
@@ -143,6 +173,7 @@ public void run() {
                 }
 
                 // Query.
+                if (publisherStopped) break;  // Check again to avoid running the query if possible
                 List<T> result = query.find();
 
                 // Notify observer(s).
@@ -160,6 +191,9 @@ public void run() {
         } finally {
             // Re-set if wrapped code throws, otherwise this publisher can no longer publish.
             publisherRunning = false;
+            synchronized (this) {
+                this.notifyAll();
+            }
         }
     }