Merge branch 'gh-793-ensure-publish-order' into release-3.0.0-alpha1

Author: greenrobot-team

objectbox-java/src/main/java/io/objectbox/BoxStore.java

@@ -918,7 +918,7 @@ long internalHandle() {
      * Note that failed or aborted transaction do not trigger observers.
      */
     public SubscriptionBuilder<Class> subscribe() {
-        return new SubscriptionBuilder<>(objectClassPublisher, null, threadPool);
+        return new SubscriptionBuilder<>(objectClassPublisher, null);
     }
 
     @Experimental
@@ -977,7 +977,7 @@ public void setDbExceptionListener(DbExceptionListener dbExceptionListener) {
      */
     @SuppressWarnings("unchecked")
     public <T> SubscriptionBuilder<Class<T>> subscribe(Class<T> forClass) {
-        return new SubscriptionBuilder<>((DataPublisher) objectClassPublisher, forClass, threadPool);
+        return new SubscriptionBuilder<>((DataPublisher) objectClassPublisher, forClass);
     }
 
     @Internal

objectbox-java/src/main/java/io/objectbox/ObjectClassPublisher.java

@@ -31,12 +31,28 @@
 import io.objectbox.reactive.DataObserver;
 import io.objectbox.reactive.DataPublisher;
 import io.objectbox.reactive.DataPublisherUtils;
+import io.objectbox.reactive.SubscriptionBuilder;
 
+/**
+ * A {@link DataPublisher} that notifies {@link DataObserver}s about changes in an entity box.
+ * Publishing is requested when a subscription is {@link SubscriptionBuilder#observer(DataObserver) observed} and
+ * then by {@link BoxStore} for each {@link BoxStore#txCommitted(Transaction, int[]) txCommitted}.
+ * Publish requests are processed on a single thread, one at a time, in the order publishing was requested.
+ */
+@SuppressWarnings("rawtypes")
 @Internal
 class ObjectClassPublisher implements DataPublisher<Class>, Runnable {
     final BoxStore boxStore;
     final MultimapSet<Integer, DataObserver<Class>> observersByEntityTypeId = MultimapSet.create(SetType.THREAD_SAFE);
-    final Deque<int[]> changesQueue = new ArrayDeque<>();
+    private final Deque<PublishRequest> changesQueue = new ArrayDeque<>();
+    private static class PublishRequest {
+        @Nullable private final DataObserver<Class> observer;
+        private final int[] entityTypeIds;
+        PublishRequest(@Nullable DataObserver<Class> observer, int[] entityTypeIds) {
+            this.observer = observer;
+            this.entityTypeIds = entityTypeIds;
+        }
+    }
     volatile boolean changePublisherRunning;
 
     ObjectClassPublisher(BoxStore boxStore) {
@@ -76,21 +92,19 @@ private void unsubscribe(DataObserver<Class> observer, int entityTypeId) {
     }
 
     @Override
-    public void publishSingle(final DataObserver<Class> observer, @Nullable final Object forClass) {
-        boxStore.internalScheduleThread(new Runnable() {
-            @Override
-            public void run() {
-                Collection<Class> entityClasses = forClass != null ? Collections.singletonList((Class) forClass) :
-                        boxStore.getAllEntityClasses();
-                for (Class entityClass : entityClasses) {
-                    try {
-                        observer.onData(entityClass);
-                    } catch (RuntimeException e) {
-                        handleObserverException(entityClass);
-                    }
-                }
+    public void publishSingle(DataObserver<Class> observer, @Nullable Object forClass) {
+        int[] entityTypeIds = forClass != null
+                ? new int[]{boxStore.getEntityTypeIdOrThrow((Class) forClass)}
+                : boxStore.getAllEntityTypeIds();
+
+        synchronized (changesQueue) {
+            changesQueue.add(new PublishRequest(observer, entityTypeIds));
+            // Only one thread at a time.
+            if (!changePublisherRunning) {
+                changePublisherRunning = true;
+                boxStore.internalScheduleThread(this);
             }
-        });
+        }
     }
 
     private void handleObserverException(Class objectClass) {
@@ -107,39 +121,50 @@ private void handleObserverException(Class objectClass) {
      */
     void publish(int[] entityTypeIdsAffected) {
         synchronized (changesQueue) {
-            changesQueue.add(entityTypeIdsAffected);
-            // Only one thread at a time
+            changesQueue.add(new PublishRequest(null, entityTypeIdsAffected));
+            // Only one thread at a time.
             if (!changePublisherRunning) {
                 changePublisherRunning = true;
                 boxStore.internalScheduleThread(this);
             }
         }
     }
 
+    /**
+     * Processes publish requests using a single thread to prevent any data generated by observers to get stale.
+     * This publisher on its own can NOT deliver stale data (the entity class types do not change).
+     * However, a {@link DataObserver} of this publisher might apply a {@link io.objectbox.reactive.DataTransformer}
+     * which queries for data which CAN get stale if delivered out of order.
+     */
     @Override
     public void run() {
         try {
             while (true) {
-                // We do not join all available array, just in case the app relies on a specific order
-                int[] entityTypeIdsAffected;
+                PublishRequest request;
                 synchronized (changesQueue) {
-                    entityTypeIdsAffected = changesQueue.pollFirst();
-                    if (entityTypeIdsAffected == null) {
+                    request = changesQueue.pollFirst();
+                    if (request == null) {
                         changePublisherRunning = false;
                         break;
                     }
                 }
-                for (int entityTypeId : entityTypeIdsAffected) {
-                    Collection<DataObserver<Class>> observers = observersByEntityTypeId.get(entityTypeId);
-                    if (observers != null && !observers.isEmpty()) {
-                        Class objectClass = boxStore.getEntityClassOrThrow(entityTypeId);
-                        try {
-                            for (DataObserver<Class> observer : observers) {
-                                observer.onData(objectClass);
-                            }
-                        } catch (RuntimeException e) {
-                            handleObserverException(objectClass);
+
+                for (int entityTypeId : request.entityTypeIds) {
+                    // If no specific observer specified, notify all current observers.
+                    Collection<DataObserver<Class>> observers = request.observer != null
+                            ? Collections.singletonList(request.observer)
+                            : observersByEntityTypeId.get(entityTypeId);
+                    if (observers == null || observers.isEmpty()) {
+                        continue; // No observers for this entity type.
+                    }
+
+                    Class entityClass = boxStore.getEntityClassOrThrow(entityTypeId);
+                    try {
+                        for (DataObserver<Class> observer : observers) {
+                            observer.onData(entityClass);
                         }
+                    } catch (RuntimeException e) {
+                        handleObserverException(entityClass);
                     }
                 }
             }

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

@@ -634,7 +634,7 @@ public Long call(long cursorHandle) {
      * it may be GCed and observers may become stale (won't receive anymore data).
      */
     public SubscriptionBuilder<List<T>> subscribe() {
-        return new SubscriptionBuilder<>(publisher, null, box.getStore().internalThreadPool());
+        return new SubscriptionBuilder<>(publisher, null);
     }
 
     /**

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

@@ -16,6 +16,9 @@
 
 package io.objectbox.query;
 
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
@@ -29,13 +32,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 SubscribedObservers<T> implements DataObserver<List<T>> {
+        @Override
+        public void onData(List<T> data) {
+        }
+    }
+    /** Placeholder observer if subscribed observers should be notified. */
+    private final SubscribedObservers<T> SUBSCRIBED_OBSERVERS = new SubscribedObservers<>();
 
     private DataObserver<Class<T>> objectClassObserver;
     private DataSubscription objectClassSubscription;
@@ -76,26 +97,73 @@ 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(SUBSCRIBED_OBSERVERS);
+            if (!publisherRunning) {
+                publisherRunning = true;
+                box.getStore().internalScheduleThread(this);
+            }
+        }
+    }
+
+    /**
+     * Processes publish requests for this query on a single thread to prevent
+     * older query results getting delivered after newer query results.
+     * To speed up processing each loop publishes to all queued observers instead of just the next in line.
+     * This reduces time spent querying and waiting for DataObserver.onData() and their potential DataTransformers.
+     */
+    @Override
+    public void run() {
+        try {
+            while (true) {
+                // Get all queued observer(s), stop processing if none.
+                List<DataObserver<List<T>>> singlePublishObservers = new ArrayList<>();
+                boolean notifySubscribedObservers = false;
+                synchronized (publishQueue) {
+                    DataObserver<List<T>> nextObserver;
+                    while ((nextObserver = publishQueue.poll()) != null) {
+                        if (SUBSCRIBED_OBSERVERS.equals(nextObserver)) {
+                            notifySubscribedObservers = true;
+                        } else {
+                            singlePublishObservers.add(nextObserver);
+                        }
+                    }
+                    if (!notifySubscribedObservers && singlePublishObservers.isEmpty()) {
+                        publisherRunning = false;
+                        break; // Stop.
+                    }
+                }
+
+                // Query.
                 List<T> result = query.find();
-                for (DataObserver<List<T>> observer : observers) {
+
+                // Notify observer(s).
+                for (DataObserver<List<T>> observer : singlePublishObservers) {
                     observer.onData(result);
                 }
+                if (notifySubscribedObservers) {
+                    // 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);
+                    }
+                }
             }
-        });
+        } finally {
+            // Re-set if wrapped code throws, otherwise this publisher can no longer publish.
+            publisherRunning = false;
+        }
     }
 
     @Override

objectbox-java/src/main/java/io/objectbox/reactive/DataTransformer.java

@@ -16,17 +16,16 @@
 
 package io.objectbox.reactive;
 
-import javax.annotation.Nullable;
-
 /**
  * Transforms or processes data before it is given to subscribed {@link DataObserver}s. A transformer is set via
  * {@link SubscriptionBuilder#transform(DataTransformer)}.
- *
+ * <p>
  * Note that a transformer is not required to actually "transform" any data.
  * Technically, it's fine to return the same data it received and just do some processing with it.
- *
- * Threading notes: Note that the transformer is always executed asynchronously.
- * It is OK to perform long lasting operations.
+ * <p>
+ * Threading notes: transformations are executed sequentially on a background thread
+ * owned by the subscription publisher. It is OK to perform long lasting operations,
+ * however this will block notifications to all other observers until finished.
  *
  * @param <FROM> Data type this transformer receives
  * @param <TO> Type of transformed data