emulate waiting for all shards before emitting Oplog updates

artificial delay avoids 100% CPU usage when starting multiple ChangeStreams

Author: jmoghisi

core/src/main/java/de/bwaldvogel/mongo/oplog/OplogCursor.java

@@ -1,6 +1,7 @@
 package de.bwaldvogel.mongo.oplog;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -27,6 +28,8 @@ public boolean isEmpty() {
 
     @Override
     public List<Document> takeDocuments(int numberToReturn) {
+        emulateWaitingForAllShards();
+
         Stream<Document> stream = oplogStream.apply(position);
 
         if (numberToReturn > 0) {
@@ -38,6 +41,16 @@ public List<Document> takeDocuments(int numberToReturn) {
         return documents;
     }
 
+    private void emulateWaitingForAllShards() {
+        try {
+            // artificial delay to avoid 100% CPU usage when starting multiple ChangeStreams
+            // emulates real ChangeStream behaviour of waiting for all shards to provide data
+            TimeUnit.MILLISECONDS.sleep(100);
+        } catch (InterruptedException e) {
+            // ignore
+        }
+    }
+
     OplogPosition getPosition() {
         return position;
     }

test-common/src/main/java/de/bwaldvogel/mongo/backend/AbstractOplogTest.java

@@ -464,7 +464,6 @@ private static <T> T getSingleValue(TestSubscriber<T> subscriber) {
     }
 
     @Test
-    @Disabled
     public void testMultipleChangeStreams() throws InterruptedException {
         Flowable.fromPublisher(asyncCollection.insertOne(json("_id: 1")))
             .test().awaitDone(5, TimeUnit.SECONDS).assertComplete();