AsyncCommandBatchCursor now uses a ResourceManager

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

@@ -188,7 +188,7 @@ private void resumeableOperation(final AsyncBlock asyncBlock, final SingleResult
                 try {
                     List<T> convertedResults;
                     try {
-                        convertedResults = convertAndProduceLastId(result, changeStreamOperation.getDecoder(),
+                        convertedResults = convertAndProduceLastId(assertNotNull(result), changeStreamOperation.getDecoder(),
                                 lastId -> resumeToken = lastId);
                     } finally {
                         cachePostBatchResumeToken(wrappedCursor);

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

@@ -165,20 +165,20 @@ private void cachePostBatchResumeToken(final AggregateResponseBatchCursor<RawBso
      * @param lastIdConsumer Is {@linkplain Consumer#accept(Object) called} iff {@code rawDocuments} is successfully converted
      *                       and the returned {@link List} is neither {@code null} nor {@linkplain List#isEmpty() empty}.
      */
-    @Nullable
     static <T> List<T> convertAndProduceLastId(@Nullable final List<RawBsonDocument> rawDocuments,
                                                final Decoder<T> decoder,
                                                final Consumer<BsonDocument> lastIdConsumer) {
-        List<T> results = null;
+        List<T> results = new ArrayList<>();
         if (rawDocuments != null) {
-            results = new ArrayList<>();
             for (RawBsonDocument rawDocument : rawDocuments) {
                 if (!rawDocument.containsKey("_id")) {
                     throw new MongoChangeStreamException("Cannot provide resume functionality when the resume token is missing.");
                 }
                 results.add(rawDocument.decode(decoder));
             }
-            lastIdConsumer.accept(rawDocuments.get(rawDocuments.size() - 1).getDocument("_id"));
+            if (!rawDocuments.isEmpty()) {
+                lastIdConsumer.accept(rawDocuments.get(rawDocuments.size() - 1).getDocument("_id"));
+            }
         }
         return results;
     }

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/internal/operation/AsyncCommandBatchCursorFunctionalSpecification.groovy

@@ -378,7 +378,7 @@ class AsyncCommandBatchCursorFunctionalSpecification extends OperationFunctional
         nextBatch().size() == 3
         nextBatch().size() == 2
         nextBatch().size() == 1
-        !nextBatch()
+        cursor.isClosed()
     }
 
     @IgnoreIf({ isSharded() })

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

@@ -86,7 +86,7 @@ class ListCollectionsOperationSpecification extends OperationFunctionalSpecifica
         cursor.next(callback)
 
         then:
-        callback.get() == null
+        callback.get() == []
 
         cleanup:
         collectionHelper.dropDatabase(madeUpDatabase)
@@ -382,7 +382,7 @@ class ListCollectionsOperationSpecification extends OperationFunctionalSpecifica
         cursor.getBatchSize() == 2
 
         cleanup:
-        consumeAsyncResults(cursor)
+        cursor?.close()
     }
 
     @IgnoreIf({ isSharded() })

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

@@ -210,7 +210,7 @@ class ListIndexesOperationSpecification extends OperationFunctionalSpecification
         cursor.getBatchSize() == 2
 
         cleanup:
-        consumeAsyncResults(cursor)
+        cursor?.close()
     }
 
     @IgnoreIf({ isSharded() })

driver-core/src/test/unit/com/mongodb/internal/operation/AsyncChangeStreamBatchCursorSpecification.groovy

@@ -46,7 +46,7 @@ class AsyncChangeStreamBatchCursorSpecification extends Specification {
         cursor.next(callback)
 
         then:
-        1 * wrapped.next(_) >> { it[0].onResult(null, null) }
+        1 * wrapped.next(_) >> { it[0].onResult([], null) }
 
         when:
         cursor.close()
@@ -78,7 +78,7 @@ class AsyncChangeStreamBatchCursorSpecification extends Specification {
         1 * wrapped.next(_) >> {
             // Simulate the user calling close while wrapped.next() is in flight
             cursor.close()
-            it[0].onResult(null, null)
+            it[0].onResult([], null)
         }
 
         then:

driver-core/src/test/unit/com/mongodb/internal/operation/AsyncCommandBatchCursorSpecification.groovy

@@ -41,7 +41,7 @@ import spock.lang.Specification
 
 import static OperationUnitSpecification.getMaxWireVersionForServerVersion
 import static com.mongodb.ReadPreference.primary
-
+import static com.mongodb.internal.operation.CommandBatchCursorHelper.MESSAGE_IF_CLOSED_AS_CURSOR
 
 class AsyncCommandBatchCursorSpecification extends Specification {
 
@@ -50,7 +50,8 @@ class AsyncCommandBatchCursorSpecification extends Specification {
         def connection = referenceCountedAsyncConnection()
         def connectionSource = getAsyncConnectionSource(connection)
 
-        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, createCommandResult([], 42), 0, batchSize, maxTimeMS, CODEC,
+        def firstBatch = createCommandResult([])
+        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, firstBatch, 0, batchSize, maxTimeMS, CODEC,
                 null, connectionSource, connection)
         def expectedCommand = new BsonDocument('getMore': new BsonInt64(CURSOR_ID))
                 .append('collection', new BsonString(NAMESPACE.getCollectionName()))
@@ -70,7 +71,7 @@ class AsyncCommandBatchCursorSpecification extends Specification {
         1 * connection.commandAsync(NAMESPACE.getDatabaseName(), expectedCommand, *_) >> {
             it.last().onResult(reply, null)
         }
-        batch == null
+        batch.isEmpty()
 
         then:
         !cursor.isClosed()
@@ -121,7 +122,8 @@ class AsyncCommandBatchCursorSpecification extends Specification {
         def connectionSource = getAsyncConnectionSource(connection)
 
         when:
-        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, createCommandResult(FIRST_BATCH, 0), 0, 0, 0, CODEC,
+        def firstBatch = createCommandResult(FIRST_BATCH, 0)
+        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, firstBatch, 0, 0, 0, CODEC,
                 null, connectionSource, connection)
 
         then:
@@ -138,7 +140,7 @@ class AsyncCommandBatchCursorSpecification extends Specification {
 
         then:
         def exception = thrown(MongoException)
-        exception.getMessage() == 'next() called after the cursor was closed.'
+        exception.getMessage() == MESSAGE_IF_CLOSED_AS_CURSOR
     }
 
     def 'should respect the limit'() {
@@ -153,7 +155,7 @@ class AsyncCommandBatchCursorSpecification extends Specification {
         def thirdBatch = [new Document('_id', 7)]
 
         when:
-        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, createCommandResult(firstBatch, 42), 7, 3, 0, CODEC,
+        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, createCommandResult(firstBatch), 7, 3, 0, CODEC,
                 null, connectionSource, connectionA)
         def batch = nextBatch(cursor)
 
@@ -178,20 +180,14 @@ class AsyncCommandBatchCursorSpecification extends Specification {
         1 * connectionB.commandAsync(*_) >> {
             connectionB.getCount() == 1
             connectionSource.getCount() == 1
-            it.last().onResult(getMoreResponse(thirdBatch, 0), null)
+                it.last().onResult(getMoreResponse(thirdBatch, 0), null)
         }
 
         then:
         batch == thirdBatch
         connectionB.getCount() == 0
         connectionSource.getCount() == 0
-
-        when:
-        batch = nextBatch(cursor)
-
-        then:
-        batch == null
-        connectionSource.getCount() == 0
+        cursor.isClosed()
     }
 
 
@@ -223,23 +219,25 @@ class AsyncCommandBatchCursorSpecification extends Specification {
 
     def 'should handle getMore when there are empty results but there is a cursor'() {
         given:
-        def connection = referenceCountedAsyncConnection(serverVersion)
-        def connectionSource = getAsyncConnectionSource(connection)
+        def connectionA = referenceCountedAsyncConnection(serverVersion, 'connectionA')
+        def connectionB = referenceCountedAsyncConnection(serverVersion, 'connectionB')
+        def connectionSource = getAsyncConnectionSource(connectionA, connectionB)
 
         when:
         def firstBatch = createCommandResult([], CURSOR_ID)
-        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, firstBatch, 3, 0, 0, CODEC, null, connectionSource, connection)
+        def cursor = new AsyncCommandBatchCursor<Document>(SERVER_ADDRESS, firstBatch, 3, 0, 0, CODEC,
+                null, connectionSource, connectionA)
         def batch = nextBatch(cursor)
 
         then:
-        1 * connection.commandAsync(*_) >> {
-            connection.getCount() == 1
+        1 * connectionA.commandAsync(*_) >> {
+            connectionA.getCount() == 1
             connectionSource.getCount() == 1
             it.last().onResult(response, null)
         }
 
-        1 * connection.commandAsync(*_) >> {
-            connection.getCount() == 1
+        1 * connectionB.commandAsync(*_) >> {
+            connectionB.getCount() == 1
             connectionSource.getCount() == 1
             it.last().onResult(response2, null)
         }
@@ -248,14 +246,16 @@ class AsyncCommandBatchCursorSpecification extends Specification {
         batch == SECOND_BATCH
 
         then:
-        connection.getCount() == 0
+        connectionA.getCount() == 0
+        connectionB.getCount() == 0
         connectionSource.getCount() == 0
 
         when:
         cursor.close()
 
         then:
-        0 * connection._
+        0 * connectionA._
+        0 * connectionB._
         connectionSource.getCount() == 0
 
         where:
@@ -338,13 +338,15 @@ class AsyncCommandBatchCursorSpecification extends Specification {
 
         then:
         connectionA.getCount() == 0
+        connectionB.getCount() == 0
+        connectionSource.getCount() == 0
         cursor.isClosed()
 
         where:
         response               | serverType               | numberOfInvocations
         getMoreResponse([])    | ServerType.LOAD_BALANCER | 2
         getMoreResponse([], 0) | ServerType.LOAD_BALANCER | 1
-        getMoreResponse([])    | ServerType.STANDALONE    | 1
+        getMoreResponse([])    | ServerType.STANDALONE    | 2
         getMoreResponse([], 0) | ServerType.STANDALONE    | 1
     }