Unify Async & Sync CommandBatchCursor testing

Part of a wider Command Cursor refactoring

JAVA-5159

Author: rozza

config/codenarc/codenarc.xml

@@ -34,12 +34,6 @@
         <exclude name="ComparisonWithSelf"/>
     </ruleset-ref>
     <ruleset-ref path='rulesets/braces.xml'/>
-    <ruleset-ref path='rulesets/concurrency.xml'>
-        <rule-config name='BusyWait'>
-            <property name='doNotApplyToFileNames' value='AsyncCommandBatchCursorFunctionalSpecification.groovy'/>
-        </rule-config>
-
-    </ruleset-ref>
     <ruleset-ref path='rulesets/convention.xml'>
         <rule-config name='NoDef'>
             <property name='doNotApplyToFilesMatching' value='.*Specification.groovy'/>

driver-core/src/main/com/mongodb/internal/operation/CommandBatchCursor.java

@@ -25,6 +25,7 @@
 import com.mongodb.ServerCursor;
 import com.mongodb.annotations.ThreadSafe;
 import com.mongodb.connection.ServerType;
+import com.mongodb.internal.VisibleForTesting;
 import com.mongodb.internal.binding.ConnectionSource;
 import com.mongodb.internal.connection.Connection;
 import com.mongodb.internal.diagnostics.logging.Logger;
@@ -54,6 +55,7 @@
 import static com.mongodb.assertions.Assertions.fail;
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.internal.Locks.withLock;
+import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
 import static com.mongodb.internal.operation.CommandBatchCursorHelper.FIRST_BATCH;
 import static com.mongodb.internal.operation.CommandBatchCursorHelper.MESSAGE_IF_CLOSED_AS_CURSOR;
 import static com.mongodb.internal.operation.CommandBatchCursorHelper.MESSAGE_IF_CLOSED_AS_ITERATOR;
@@ -69,18 +71,19 @@ class CommandBatchCursor<T> implements AggregateResponseBatchCursor<T> {
 
     private final MongoNamespace namespace;
     private final int limit;
-    private final Decoder<T> decoder;
     private final long maxTimeMS;
-    private int batchSize;
+    private final Decoder<T> decoder;
     @Nullable
     private final BsonValue comment;
+    private final int maxWireVersion;
+    private final boolean firstBatchEmpty;
+    private final ResourceManager resourceManager;
+
+    private int batchSize;
     private CommandCursorResult<T> commandCursorResult;
+    private int count;
     @Nullable
     private List<T> nextBatch;
-    private int count;
-    private final boolean firstBatchEmpty;
-    private final int maxWireVersion;
-    private final ResourceManager resourceManager;
 
     CommandBatchCursor(
             final BsonDocument commandCursorDocument,
@@ -162,6 +165,11 @@ private List<T> doNext() {
         return retVal;
     }
 
+    @VisibleForTesting(otherwise = PRIVATE)
+    boolean isClosed() {
+        return !resourceManager.operable();
+    }
+
     @Override
     public void setBatchSize(final int batchSize) {
         this.batchSize = batchSize;

driver-core/src/test/functional/com/mongodb/OperationFunctionalSpecification.groovy

@@ -207,13 +207,6 @@ class OperationFunctionalSpecification extends Specification {
         }
     }
 
-    def consumeAsyncResults(cursor) {
-        def batch = next(cursor, true)
-        while (batch != null) {
-            batch = next(cursor, true)
-        }
-    }
-
     void testOperation(Map params) {
         params.async = params.async != null ? params.async : false
         params.result = params.result != null ? params.result : null

driver-core/src/test/functional/com/mongodb/client/model/OperationTest.java

@@ -18,14 +18,17 @@
 
 import com.mongodb.ClusterFixture;
 import com.mongodb.MongoNamespace;
+import com.mongodb.async.FutureResultCallback;
 import com.mongodb.client.test.CollectionHelper;
 import com.mongodb.internal.connection.ServerHelper;
+import com.mongodb.internal.validator.NoOpFieldNameValidator;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
 import org.bson.BsonDouble;
 import org.bson.BsonValue;
 import org.bson.Document;
+import org.bson.FieldNameValidator;
 import org.bson.codecs.BsonDocumentCodec;
 import org.bson.codecs.DecoderContext;
 import org.bson.codecs.DocumentCodec;
@@ -39,8 +42,11 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import static com.mongodb.ClusterFixture.TIMEOUT;
 import static com.mongodb.ClusterFixture.checkReferenceCountReachesTarget;
 import static com.mongodb.ClusterFixture.getAsyncBinding;
 import static com.mongodb.ClusterFixture.getBinding;
@@ -50,14 +56,17 @@
 import static com.mongodb.client.model.Aggregates.sort;
 import static java.util.stream.Collectors.toList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 public abstract class OperationTest {
 
     protected static final DocumentCodec DOCUMENT_DECODER = new DocumentCodec();
+    protected static final FieldNameValidator NO_OP_FIELD_NAME_VALIDATOR = new NoOpFieldNameValidator();
 
     @BeforeEach
     public void beforeEach() {
-        ServerHelper.checkPool(getPrimary());
+        assumeTrue(ServerHelper.checkPoolCount(getPrimary()) == 0, "Sync Pool count not zero");
+        assumeTrue(ServerHelper.checkAsyncPoolCount(getPrimary()) == 0, "Async Pool count not zero");
         CollectionHelper.drop(getNamespace());
     }
 
@@ -77,15 +86,15 @@ private CollectionHelper<BsonDocument> getCollectionHelper(final MongoNamespace
         return new CollectionHelper<>(new BsonDocumentCodec(), namespace);
     }
 
-    private String getDatabaseName() {
+    protected String getDatabaseName() {
         return ClusterFixture.getDefaultDatabaseName();
     }
 
-    private String getCollectionName() {
+    protected String getCollectionName() {
         return "test";
     }
 
-    MongoNamespace getNamespace() {
+    protected MongoNamespace getNamespace() {
         return new MongoNamespace(getDatabaseName(), getCollectionName());
     }
 
@@ -97,7 +106,6 @@ public static BsonDocument toBsonDocument(final BsonDocument bsonDocument) {
         return getDefaultCodecRegistry().get(BsonDocument.class).decode(bsonDocument.asBsonReader(), DecoderContext.builder().build());
     }
 
-
     protected List<Bson> assertPipeline(final String stageAsString, final Bson stage) {
         List<Bson> pipeline = Collections.singletonList(stage);
         return assertPipeline(stageAsString, pipeline);
@@ -159,4 +167,25 @@ protected List<Object> aggregateWithWindowFields(@Nullable final Object partitio
                 .map(doc -> doc.get("result"))
                 .collect(toList());
     }
+
+    protected <T> void ifNotNull(@Nullable final T maybeNull, final Consumer<T> consumer) {
+        if (maybeNull != null) {
+            consumer.accept(maybeNull);
+        }
+    }
+
+    protected void sleep(final long ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        }
+    }
+
+    protected <T> T block(final Consumer<FutureResultCallback<T>> consumer) {
+        FutureResultCallback<T> cb = new FutureResultCallback<>();
+        consumer.accept(cb);
+        return cb.get(TIMEOUT, TimeUnit.SECONDS);
+    }
 }

driver-core/src/test/functional/com/mongodb/internal/connection/ServerHelper.java

@@ -36,6 +36,14 @@ public static void checkPool(final ServerAddress address) {
         checkPool(address, getAsyncCluster());
     }
 
+    public static int checkPoolCount(final ServerAddress address) {
+        return getConnectionPool(address, getCluster()).getInUseCount();
+    }
+
+    public static int checkAsyncPoolCount(final ServerAddress address) {
+        return getConnectionPool(address, getAsyncCluster()).getInUseCount();
+    }
+
     public static void waitForLastRelease(final Cluster cluster) {
         for (ServerDescription cur : cluster.getCurrentDescription().getServerDescriptions()) {
             if (cur.isOk()) {
@@ -45,12 +53,11 @@ public static void waitForLastRelease(final Cluster cluster) {
     }
 
     public static void waitForLastRelease(final ServerAddress address, final Cluster cluster) {
-        ConcurrentPool<UsageTrackingInternalConnection> pool = connectionPool(
-                cluster.selectServer(new ServerAddressSelector(address), OPERATION_CONTEXT).getServer());
+        ConcurrentPool<UsageTrackingInternalConnection> pool = getConnectionPool(address, cluster);
         long startTime = System.currentTimeMillis();
         while (pool.getInUseCount() > 0) {
             try {
-                sleep(10);
+                sleep(100);
                 if (System.currentTimeMillis() > startTime + ClusterFixture.TIMEOUT * 1000) {
                     throw new MongoTimeoutException("Timed out waiting for pool in use count to drop to 0.  Now at: "
                                                             + pool.getInUseCount());
@@ -61,11 +68,15 @@ public static void waitForLastRelease(final ServerAddress address, final Cluster
         }
     }
 
+    private static ConcurrentPool<UsageTrackingInternalConnection> getConnectionPool(final ServerAddress address, final Cluster cluster) {
+        return connectionPool(cluster.selectServer(new ServerAddressSelector(address), OPERATION_CONTEXT).getServer());
+    }
+
     private static void checkPool(final ServerAddress address, final Cluster cluster) {
-        ConcurrentPool<UsageTrackingInternalConnection> pool = connectionPool(
-                cluster.selectServer(new ServerAddressSelector(address), OPERATION_CONTEXT).getServer());
-        if (pool.getInUseCount() > 0) {
-            throw new IllegalStateException("Connection pool in use count is " + pool.getInUseCount());
+        try {
+            waitForLastRelease(address, cluster);
+        } catch (MongoTimeoutException e) {
+            throw new IllegalStateException(e.getMessage());
         }
     }
 

driver-core/src/test/functional/com/mongodb/internal/operation/AggregateOperationSpecification.groovy

@@ -50,7 +50,7 @@ import org.bson.codecs.BsonDocumentCodec
 import org.bson.codecs.DocumentCodec
 import spock.lang.IgnoreIf
 
-import static QueryOperationHelper.getKeyPattern
+import static TestOperationHelper.getKeyPattern
 import static com.mongodb.ClusterFixture.OPERATION_CONTEXT
 import static com.mongodb.ClusterFixture.TIMEOUT_SETTINGS
 import static com.mongodb.ClusterFixture.TIMEOUT_SETTINGS_WITH_MAX_TIME