QueryPublisher: queue publish requests to ensure delivery order matches.

Author: greenrobot-team

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

@@ -16,6 +16,8 @@
 
 package io.objectbox.query;
 
+import java.util.ArrayDeque;
+import java.util.Deque;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
@@ -29,13 +31,31 @@
 import io.objectbox.reactive.DataPublisher;
 import io.objectbox.reactive.DataPublisherUtils;
 import io.objectbox.reactive.DataSubscription;
+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.
+ */
 @Internal
-class QueryPublisher<T> implements DataPublisher<List<T>> {
+class QueryPublisher<T> implements DataPublisher<List<T>>, Runnable {
 
     private final Query<T> query;
     private final Box<T> box;
     private final Set<DataObserver<List<T>>> observers = new CopyOnWriteArraySet<>();
+    private final Deque<DataObserver<List<T>>> publishQueue = new ArrayDeque<>();
+    private volatile boolean publisherRunning = false;
+
+    private static class AllObservers<T> implements DataObserver<List<T>> {
+        @Override
+        public void onData(List<T> data) {
+        }
+    }
+    /** Placeholder observer if all observers should be notified. */
+    private final AllObservers<T> ALL_OBSERVERS = new AllObservers<>();
 
     private DataObserver<Class<T>> objectClassObserver;
     private DataSubscription objectClassSubscription;
@@ -76,26 +96,60 @@ public void onData(Class<T> objectClass) {
     }
 
     @Override
-    public void publishSingle(final DataObserver<List<T>> observer, @Nullable Object param) {
-        box.getStore().internalScheduleThread(new Runnable() {
-            @Override
-            public void run() {
-                List<T> result = query.find();
-                observer.onData(result);
+    public void publishSingle(DataObserver<List<T>> observer, @Nullable Object param) {
+        synchronized (publishQueue) {
+            publishQueue.add(observer);
+            if (!publisherRunning) {
+                publisherRunning = true;
+                box.getStore().internalScheduleThread(this);
             }
-        });
+        }
     }
 
     void publish() {
-        box.getStore().internalScheduleThread(new Runnable() {
-            @Override
-            public void run() {
+        synchronized (publishQueue) {
+            publishQueue.add(ALL_OBSERVERS);
+            if (!publisherRunning) {
+                publisherRunning = true;
+                box.getStore().internalScheduleThread(this);
+            }
+        }
+    }
+
+    @Override
+    public void run() {
+        /*
+         * Process publish requests for this query on a single thread to avoid an older request
+         * racing a new one (and causing outdated results to be delivered last).
+         */
+        try {
+            while (true) {
+                // Get next observer(s).
+                DataObserver<List<T>> observer;
+                synchronized (publishQueue) {
+                    observer = publishQueue.pollFirst();
+                    if (observer == null) {
+                        publisherRunning = false;
+                        break;
+                    }
+                }
+
+                // Query, then notify observer(s).
                 List<T> result = query.find();
-                for (DataObserver<List<T>> observer : observers) {
+                if (ALL_OBSERVERS.equals(observer)) {
+                    // Use current list of observers to avoid notifying unsubscribed observers.
+                    Set<DataObserver<List<T>>> observers = this.observers;
+                    for (DataObserver<List<T>> dataObserver : observers) {
+                        dataObserver.onData(result);
+                    }
+                } else {
                     observer.onData(result);
                 }
             }
-        });
+        } finally {
+            // Re-set if wrapped code throws, otherwise this publisher can no longer publish.
+            publisherRunning = false;
+        }
     }
 
     @Override