To deliver in order run transform on same thread as ActionObserver was notified on.

Author: greenrobot-team

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

@@ -1064,7 +1064,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
@@ -1159,7 +1159,7 @@ public void setDbExceptionListener(@Nullable DbExceptionListener dbExceptionList
      */
     @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/query/Query.java

@@ -611,7 +611,7 @@ public long remove() {
      * 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/reactive/SubscriptionBuilder.java

@@ -16,8 +16,6 @@
 
 package io.objectbox.reactive;
 
-import java.util.concurrent.ExecutorService;
-
 import javax.annotation.Nullable;
 
 import io.objectbox.annotation.apihint.Internal;
@@ -44,7 +42,6 @@
 public class SubscriptionBuilder<T> {
     private final DataPublisher<T> publisher;
     private final Object publisherParam;
-    private final ExecutorService threadPool;
     private DataObserver<T> observer;
     //    private Runnable firstRunnable;
     private boolean weak;
@@ -58,10 +55,9 @@ public class SubscriptionBuilder<T> {
 
 
     @Internal
-    public SubscriptionBuilder(DataPublisher<T> publisher, @Nullable Object param, ExecutorService threadPool) {
+    public SubscriptionBuilder(DataPublisher<T> publisher, @Nullable Object param) {
         this.publisher = publisher;
         publisherParam = param;
-        this.threadPool = threadPool;
     }
 
     //    public Observable<T> runFirst(Runnable firstRunnable) {
@@ -214,19 +210,27 @@ public void onData(final T data) {
             }
         }
 
+        /**
+         * Runs on the thread of the {@link #onData(Object)} caller to ensure data is delivered
+         * in the same order as {@link #onData(Object)} was called, to prevent delivering stale data.
+         * <p>
+         * For both ObjectClassPublisher and QueryPublisher this is the asynchronous
+         * thread publish requests are processed on.
+         * <p>
+         * This could be optimized in the future to allow parallel execution,
+         * but this would require an ordering mechanism for the transformed data.
+         */
         private void transformAndContinue(final T data) {
-            threadPool.submit(() -> {
-                if (subscription.isCanceled()) {
-                    return;
-                }
-                try {
-                    // Type erasure FTW
-                    T result = (T) transformer.transform(data);
-                    callOnData(result);
-                } catch (Throwable th) {
-                    callOnError(th, "Transformer failed without an ErrorObserver set");
-                }
-            });
+            if (subscription.isCanceled()) {
+                return;
+            }
+            try {
+                // Type erasure FTW
+                T result = (T) transformer.transform(data);
+                callOnData(result);
+            } catch (Throwable th) {
+                callOnError(th, "Transformer failed without an ErrorObserver set");
+            }
         }
 
         private void callOnError(Throwable th, String msgNoErrorObserver) {

objectbox-rxjava/src/test/java/io/objectbox/query/MockQuery.java

@@ -35,7 +35,7 @@ public MockQuery(boolean hasOrder) {
         // when(box.getStore()).thenReturn(boxStore);
          query = mock(Query.class);
         fakeQueryPublisher = new FakeQueryPublisher();
-        SubscriptionBuilder subscriptionBuilder = new SubscriptionBuilder(fakeQueryPublisher, null, null);
+        SubscriptionBuilder subscriptionBuilder = new SubscriptionBuilder(fakeQueryPublisher, null);
         when(query.subscribe()).thenReturn(subscriptionBuilder);
     }
 

objectbox-rxjava3/src/test/java/io/objectbox/query/MockQuery.java

@@ -40,9 +40,7 @@ public MockQuery(boolean hasOrder) {
         //noinspection unchecked It's a unit test, casting is fine.
         query = (Query<T>) mock(Query.class);
         fakeQueryPublisher = new FakeQueryPublisher<>();
-        //noinspection ConstantConditions ExecutorService only used for transforms.
-        SubscriptionBuilder<List<T>> subscriptionBuilder = new SubscriptionBuilder<>(
-                fakeQueryPublisher, null, null);
+        SubscriptionBuilder<List<T>> subscriptionBuilder = new SubscriptionBuilder<>(fakeQueryPublisher, null);
         when(query.subscribe()).thenReturn(subscriptionBuilder);
     }