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

@@ -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/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/reactive/SubscriptionBuilder.java

@@ -16,11 +16,8 @@
 
 package io.objectbox.reactive;
 
-import java.util.concurrent.ExecutorService;
-
 import javax.annotation.Nullable;
 
-import io.objectbox.annotation.apihint.Beta;
 import io.objectbox.annotation.apihint.Internal;
 
 /**
@@ -45,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;
@@ -59,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) {
@@ -215,22 +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(new Runnable() {
-                @Override
-                public void run() {
-                    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);
     }