Test transformed data is received in order of publish.

Author: greenrobot-team

tests/objectbox-java-test/src/test/java/io/objectbox/ObjectClassObserverTest.java

@@ -23,8 +23,10 @@
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import io.objectbox.query.QueryObserverTest;
 import io.objectbox.reactive.DataObserver;
 import io.objectbox.reactive.DataSubscription;
 import io.objectbox.reactive.DataTransformer;
@@ -194,6 +196,42 @@ private void testTransform(TestScheduler scheduler) throws InterruptedException
         assertEquals(0, objectCounts.size());
     }
 
+    /**
+     * There is an identical test asserting QueryPublisher at
+     * {@link QueryObserverTest#transform_inOrderOfPublish()}.
+     */
+    @Test
+    public void transform_inOrderOfPublish() {
+        final CountDownLatch latch = new CountDownLatch(2);
+        final AtomicBoolean isLongTransform = new AtomicBoolean(true);
+        final List<Integer> placing = new CopyOnWriteArrayList<>();
+
+        // Make first transformation take longer than second.
+        DataSubscription subscription = store.subscribe(TestEntity.class).transform(source -> {
+            if (isLongTransform.compareAndSet(true, false)) {
+                // Wait long enough so publish triggered by transaction
+                // can overtake publish triggered during observer() call.
+                Thread.sleep(1000);
+                return 1; // First, during observer() call.
+            }
+            return 2; // Second, due to transaction.
+        }).observer(data -> {
+            placing.add(data);
+            latch.countDown();
+        });
+
+        // Trigger publish due to transaction.
+        store.runInTx(() -> putTestEntities(1));
+
+        assertLatchCountedDown(latch, 3);
+        subscription.cancel();
+
+        // Second publish request should still deliver second.
+        assertEquals(2, placing.size());
+        assertEquals(1, (int) placing.get(0));
+        assertEquals(2, (int) placing.get(1));
+    }
+
     @Test
     public void testScheduler() {
         TestScheduler scheduler = new TestScheduler();

tests/objectbox-java-test/src/test/java/io/objectbox/query/QueryObserverTest.java

@@ -27,6 +27,7 @@
 
 import io.objectbox.AbstractObjectBoxTest;
 import io.objectbox.Box;
+import io.objectbox.ObjectClassObserverTest;
 import io.objectbox.TestEntity;
 import io.objectbox.reactive.DataObserver;
 import io.objectbox.reactive.DataSubscription;
@@ -149,6 +150,44 @@ public void testTransformer() throws InterruptedException {
         assertEquals(2003 + 2007 + 2002, (int) receivedSums.get(1));
     }
 
+    /**
+     * There is an identical test asserting ObjectClassPublisher at
+     * {@link ObjectClassObserverTest#transform_inOrderOfPublish()}.
+     */
+    @Test
+    public void transform_inOrderOfPublish() {
+        Query<TestEntity> query = box.query().build();
+
+        final CountDownLatch latch = new CountDownLatch(2);
+        final AtomicBoolean isLongTransform = new AtomicBoolean(true);
+        final List<Integer> placing = new CopyOnWriteArrayList<>();
+
+        // Make first transformation take longer than second.
+        DataSubscription subscription = query.subscribe().transform(source -> {
+            if (isLongTransform.compareAndSet(true, false)) {
+                // Wait long enough so publish triggered by transaction
+                // can overtake publish triggered during observer() call.
+                Thread.sleep(1000);
+                return 1; // First, during observer() call.
+            }
+            return 2; // Second, due to transaction.
+        }).observer(data -> {
+            placing.add(data);
+            latch.countDown();
+        });
+
+        // Trigger publish due to transaction.
+        store.runInTx(() -> putTestEntities(1));
+
+        assertLatchCountedDown(latch, 3);
+        subscription.cancel();
+
+        // Second publish request should still deliver second.
+        assertEquals(2, placing.size());
+        assertEquals(1, (int) placing.get(0));
+        assertEquals(2, (int) placing.get(1));
+    }
+
     private void putTestEntitiesScalars() {
         putTestEntities(10, null, 2000);
     }