ObjectClassPublisher: process all requests on same thread to ensure order.

Author: greenrobot-team

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);
                     }
                 }
             }