Abstract Cursor Resource Manager

driver-core/src/main/com/mongodb/assertions/Assertions.java

@@ -21,6 +21,7 @@
 import com.mongodb.lang.Nullable;
 
 import java.util.Collection;
+import java.util.function.Supplier;
 
 /**
  * <p>Design by contract assertions.</p> <p>This class is not part of the public API and may be removed or changed at any time.</p>
@@ -226,6 +227,19 @@ public static AssertionError fail(final String msg) throws AssertionError {
         throw new AssertionError(assertNotNull(msg));
     }
 
+    /**
+     * @param supplier the supplier to check
+     * @return {@code supplier.get()}
+     * @throws AssertionError If {@code supplier.get()} throws an exception
+     */
+    public static <T> T doesNotThrow(final Supplier<T> supplier) throws AssertionError {
+        try {
+            return supplier.get();
+        } catch (Exception e) {
+            throw new AssertionError(e.getMessage(), e);
+        }
+    }
+
     private Assertions() {
     }
 }

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

@@ -224,16 +224,16 @@ private CommandReadTransformer<BsonDocument, CommandBatchCursor<T>> transformer(
         return (result, source, connection) -> {
             // TODO (CSOT) JAVA-4058
             long maxAwaitTimeMS = timeoutSettings.getMaxAwaitTimeMS();
-            return new CommandBatchCursor<>(result, 0, batchSize != null ? batchSize : 0, maxAwaitTimeMS, decoder,
-                    comment, source, connection);
+            return new CommandBatchCursor<>(result, batchSize != null ? batchSize : 0, maxAwaitTimeMS, decoder,
+                                            comment, source, connection);
         };
     }
 
     private CommandReadTransformerAsync<BsonDocument, AsyncBatchCursor<T>> asyncTransformer() {
         return (result, source, connection) -> {
             // TODO (CSOT) JAVA-4058
             long maxAwaitTimeMS = timeoutSettings.getMaxAwaitTimeMS();
-            return new AsyncCommandBatchCursor<>(result, 0, batchSize != null ? batchSize : 0, maxAwaitTimeMS, decoder,
+            return new AsyncCommandBatchCursor<>(result, batchSize != null ? batchSize : 0, maxAwaitTimeMS, decoder,
                     comment, source, connection);
         };
     }

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/AsyncOperationHelper.java

@@ -61,6 +61,10 @@ interface AsyncCallableWithConnection {
         void call(@Nullable AsyncConnection connection, @Nullable Throwable t);
     }
 
+    interface AsyncCallableConnectionWithCallback<T> {
+        void call(AsyncConnection connection, SingleResultCallback<T> callback);
+    }
+
     interface AsyncCallableWithSource {
         void call(@Nullable AsyncConnectionSource source, @Nullable Throwable t);
     }
@@ -335,8 +339,7 @@ static <T> CommandReadTransformerAsync<BsonDocument, AsyncBatchCursor<T>> asyncS
 
     static <T> AsyncBatchCursor<T> cursorDocumentToAsyncBatchCursor(final BsonDocument cursorDocument, final Decoder<T> decoder,
             final BsonValue comment, final AsyncConnectionSource source, final AsyncConnection connection, final int batchSize) {
-        return new AsyncCommandBatchCursor<>(cursorDocument, 0, batchSize, 0, decoder,
-                comment, source, connection);
+        return new AsyncCommandBatchCursor<>(cursorDocument, batchSize, 0, decoder, comment, source, connection);
     }
 
     static <T> SingleResultCallback<T> releasingCallback(final SingleResultCallback<T> wrapped, final AsyncConnection connection) {

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

@@ -33,6 +33,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import static com.mongodb.assertions.Assertions.assertNotNull;
 import static com.mongodb.internal.operation.ChangeStreamBatchCursorHelper.isResumableError;
 import static com.mongodb.internal.operation.SyncOperationHelper.withReadConnectionSource;
 
@@ -94,8 +95,9 @@ public int available() {
     public List<T> tryNext() {
         return resumeableOperation(commandBatchCursor -> {
             try {
-                return convertAndProduceLastId(commandBatchCursor.tryNext(), changeStreamOperation.getDecoder(),
-                        lastId -> resumeToken = lastId);
+                List<RawBsonDocument> tryNext = commandBatchCursor.tryNext();
+                return tryNext == null ? null
+                        : convertAndProduceLastId(tryNext, changeStreamOperation.getDecoder(), lastId -> resumeToken = lastId);
             } finally {
                 cachePostBatchResumeToken(commandBatchCursor);
             }
@@ -165,19 +167,17 @@ 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,
+    static <T> List<T> convertAndProduceLastId(final List<RawBsonDocument> rawDocuments,
                                                final Decoder<T> decoder,
                                                final Consumer<BsonDocument> lastIdConsumer) {
-        List<T> results = null;
-        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));
+        List<T> results = new ArrayList<>();
+        for (RawBsonDocument rawDocument : assertNotNull(rawDocuments)) {
+            if (!rawDocument.containsKey("_id")) {
+                throw new MongoChangeStreamException("Cannot provide resume functionality when the resume token is missing.");
             }
+            results.add(rawDocument.decode(decoder));
+        }
+        if (!rawDocuments.isEmpty()) {
             lastIdConsumer.accept(rawDocuments.get(rawDocuments.size() - 1).getDocument("_id"));
         }
         return results;