address Ross comments

Author: Almas Abdrazak

…iteOperationThenCursorReadOperation.java …iteOperationThenCursorReadOperation.javadriver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/VoidWriteOperationThenCursorReadOperation.java renamed to driver-core/src/main/com/mongodb/internal/operation/VoidWriteOperationThenCursorReadOperation.java driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/VoidWriteOperationThenCursorReadOperation.java renamed to driver-core/src/main/com/mongodb/internal/operation/VoidWriteOperationThenCursorReadOperation.java

@@ -14,27 +14,40 @@
  * limitations under the License.
  */
 
-package com.mongodb.reactivestreams.client.internal;
+package com.mongodb.internal.operation;
 
 import com.mongodb.MongoNamespace;
 import com.mongodb.internal.async.AsyncBatchCursor;
 import com.mongodb.internal.async.SingleResultCallback;
 import com.mongodb.internal.binding.AsyncReadWriteBinding;
+import com.mongodb.internal.binding.ReadWriteBinding;
 import com.mongodb.internal.connection.OperationContext;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
-import com.mongodb.internal.operation.ReadOperationCursor;
-import com.mongodb.internal.operation.WriteOperation;
 
-class VoidWriteOperationThenCursorReadOperation<T> implements AsyncWriteThenReadOperationCursor<T> {
+/**
+ * A {@link WriteThenReadOperationCursor} that performs a {@link WriteOperation} returning
+ * {@code Void} followed by a {@link ReadOperationCursor}, using a single read-write
+ * binding for both phases.
+ *
+ * <p>This class is not part of the public API and may be removed or changed at any time</p>
+ */
+public final class VoidWriteOperationThenCursorReadOperation<T> implements WriteThenReadOperationCursor<T> {
     private final WriteOperation<Void> writeOperation;
     private final ReadOperationCursor<T> cursorReadOperation;
 
-    VoidWriteOperationThenCursorReadOperation(final WriteOperation<Void> writeOperation,
-                                              final ReadOperationCursor<T> cursorReadOperation) {
+    public VoidWriteOperationThenCursorReadOperation(final WriteOperation<Void> writeOperation,
+                                                     final ReadOperationCursor<T> cursorReadOperation) {
         this.writeOperation = writeOperation;
         this.cursorReadOperation = cursorReadOperation;
     }
 
+    public WriteOperation<Void> getWriteOperation() {
+        return writeOperation;
+    }
+
+    public ReadOperationCursor<T> getCursorReadOperation() {
+        return cursorReadOperation;
+    }
+
     @Override
     public String getCommandName() {
         return writeOperation.getCommandName();
@@ -45,6 +58,12 @@ public MongoNamespace getNamespace() {
         return writeOperation.getNamespace();
     }
 
+    @Override
+    public BatchCursor<T> execute(final ReadWriteBinding binding, final OperationContext operationContext) {
+        writeOperation.execute(binding, operationContext);
+        return cursorReadOperation.execute(binding, operationContext);
+    }
+
     @Override
     public void executeAsync(final AsyncReadWriteBinding binding, final OperationContext operationContext,
                              final SingleResultCallback<AsyncBatchCursor<T>> callback) {

…n/AsyncWriteThenReadOperationCursor.java …ration/WriteThenReadOperationCursor.javadriver-core/src/main/com/mongodb/internal/operation/AsyncWriteThenReadOperationCursor.java renamed to driver-core/src/main/com/mongodb/internal/operation/WriteThenReadOperationCursor.java driver-core/src/main/com/mongodb/internal/operation/AsyncWriteThenReadOperationCursor.java renamed to driver-core/src/main/com/mongodb/internal/operation/WriteThenReadOperationCursor.java

@@ -20,17 +20,20 @@
 import com.mongodb.internal.async.AsyncBatchCursor;
 import com.mongodb.internal.async.SingleResultCallback;
 import com.mongodb.internal.binding.AsyncReadWriteBinding;
+import com.mongodb.internal.binding.ReadWriteBinding;
 import com.mongodb.internal.connection.OperationContext;
 
 /**
- * An async-only operation that performs a write followed by a read that returns a cursor.
+ * An operation that performs a write followed by a read that returns a cursor, using a
+ * single read-write binding for both phases.
  *
- * <p>Unlike {@link ReadOperationCursor}, this operation requires an {@link AsyncReadWriteBinding}
- * so that both the write and the read portions can be executed without narrowing casts.
+ * <p>Having a dedicated interface lets the executor hand the operation a binding that is
+ * both a {@link ReadWriteBinding} and an {@link AsyncReadWriteBinding}, avoiding the
+ * narrowing cast from a read-only binding that would otherwise be required.
  *
  * <p>This class is not part of the public API and may be removed or changed at any time</p>
  */
-public interface AsyncWriteThenReadOperationCursor<T> {
+public interface WriteThenReadOperationCursor<T> {
 
     /**
      * @return the command name of the write phase of this operation (e.g. "mapReduce", "aggregate")
@@ -42,6 +45,16 @@ public interface AsyncWriteThenReadOperationCursor<T> {
      */
     MongoNamespace getNamespace();
 
+    /**
+     * Executes the write phase followed by the read phase, returning a {@link BatchCursor}
+     * over the results of the read.
+     *
+     * @param binding the read-write binding used by both phases
+     * @param operationContext the operation context to use
+     * @return the batch cursor produced by the read phase
+     */
+    BatchCursor<T> execute(ReadWriteBinding binding, OperationContext operationContext);
+
     /**
      * Executes the write phase followed by the read phase, yielding an {@link AsyncBatchCursor}
      * over the results of the read.

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/ClientSessionPublisherImpl.java

@@ -27,11 +27,11 @@
 import com.mongodb.internal.observability.micrometer.TracingManager;
 import com.mongodb.internal.observability.micrometer.TransactionSpan;
 import com.mongodb.internal.operation.AbortTransactionOperation;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
 import com.mongodb.internal.operation.CommitTransactionOperation;
 import com.mongodb.internal.operation.ReadOperation;
 import com.mongodb.internal.operation.WriteConcernHelper;
 import com.mongodb.internal.operation.WriteOperation;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.internal.session.BaseClientSessionImpl;
 import com.mongodb.internal.session.ServerSessionPool;
 import com.mongodb.lang.Nullable;
@@ -89,7 +89,8 @@ public boolean notifyMessageSent() {
 
     @Override
     public void notifyOperationInitiated(final Object operation) {
-        assertTrue(operation instanceof ReadOperation || operation instanceof WriteOperation || operation instanceof AsyncWriteThenReadOperationCursor);
+        assertTrue(operation instanceof ReadOperation || operation instanceof WriteOperation
+                || operation instanceof WriteThenReadOperationCursor);
         if (!(hasActiveTransaction() || operation instanceof CommitTransactionOperation)) {
             assertTrue(getPinnedServerAddress() == null
                     || (transactionState != TransactionState.ABORTED && transactionState != TransactionState.NONE));

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MapReducePublisherImpl.java

@@ -34,6 +34,7 @@
 import com.mongodb.internal.operation.Operations;
 import com.mongodb.internal.operation.ReadOperation;
 import com.mongodb.internal.operation.ReadOperationCursor;
+import com.mongodb.internal.operation.VoidWriteOperationThenCursorReadOperation;
 import com.mongodb.internal.operation.WriteOperation;
 import com.mongodb.lang.Nullable;
 import com.mongodb.reactivestreams.client.ClientSession;

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MongoOperationPublisher.java

@@ -61,7 +61,7 @@
 import com.mongodb.internal.async.SingleResultCallback;
 import com.mongodb.internal.bulk.WriteRequest;
 import com.mongodb.internal.async.AsyncBatchCursor;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.internal.operation.IndexHelper;
 import com.mongodb.internal.operation.Operations;
 import com.mongodb.internal.operation.ReadOperation;
@@ -517,9 +517,9 @@ <T> Mono<T> createReadOperationMono(final Supplier<TimeoutSettings> timeoutSetti
 
     <R> Mono<AsyncBatchCursor<R>> createWriteThenReadOperationMono(
             final Function<Operations<?>, TimeoutSettings> timeoutSettingsFunction,
-            final Supplier<AsyncWriteThenReadOperationCursor<R>> operationSupplier,
+            final Supplier<WriteThenReadOperationCursor<R>> operationSupplier,
             @Nullable final ClientSession clientSession) {
-        AsyncWriteThenReadOperationCursor<R> operation = operationSupplier.get();
+        WriteThenReadOperationCursor<R> operation = operationSupplier.get();
         return getExecutor(timeoutSettingsFunction.apply(operations))
                 .execute(operation, getReadConcern(), clientSession);
     }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OperationExecutor.java

@@ -20,9 +20,9 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.internal.TimeoutSettings;
 import com.mongodb.internal.async.AsyncBatchCursor;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
 import com.mongodb.internal.operation.ReadOperation;
 import com.mongodb.internal.operation.WriteOperation;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.lang.Nullable;
 import com.mongodb.reactivestreams.client.ClientSession;
 import reactor.core.publisher.Mono;
@@ -67,7 +67,7 @@ <T> Mono<T> execute(ReadOperation<?, T> operation, ReadPreference readPreference
      * @param session the session to associate this operation with
      * @param <T> the document type returned by the cursor.
      */
-    <T> Mono<AsyncBatchCursor<T>> execute(AsyncWriteThenReadOperationCursor<T> operation,
+    <T> Mono<AsyncBatchCursor<T>> execute(WriteThenReadOperationCursor<T> operation,
             ReadConcern readConcern, @Nullable ClientSession session);
 
     /**

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OperationExecutorImpl.java

@@ -34,10 +34,10 @@
 import com.mongodb.internal.observability.micrometer.Span;
 import com.mongodb.internal.observability.micrometer.TracingManager;
 import com.mongodb.internal.async.AsyncBatchCursor;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
 import com.mongodb.internal.operation.OperationHelper;
 import com.mongodb.internal.operation.ReadOperation;
 import com.mongodb.internal.operation.WriteOperation;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.lang.Nullable;
 import com.mongodb.reactivestreams.client.ClientSession;
 import com.mongodb.reactivestreams.client.ReactiveContextProvider;
@@ -182,7 +182,7 @@ public <T> Mono<T> execute(final WriteOperation<T> operation, final ReadConcern
     }
 
     @Override
-    public <T> Mono<AsyncBatchCursor<T>> execute(final AsyncWriteThenReadOperationCursor<T> operation, final ReadConcern readConcern,
+    public <T> Mono<AsyncBatchCursor<T>> execute(final WriteThenReadOperationCursor<T> operation, final ReadConcern readConcern,
             @Nullable final ClientSession session) {
         isTrue("open", !mongoClient.getCluster().isClosed());
         notNull("operation", operation);

driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/MapReducePublisherImplTest.java

@@ -20,7 +20,7 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
 import com.mongodb.client.model.Sorts;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.internal.operation.MapReduceStatistics;
 import com.mongodb.internal.operation.MapReduceToCollectionOperation;
 import com.mongodb.internal.operation.MapReduceWithInlineResultsOperation;
@@ -66,7 +66,7 @@ void shouldRouteNonInlineMapReduceThroughWriteThenReadPath() {
 
         Flux.from(publisher).blockFirst();
 
-        AsyncWriteThenReadOperationCursor<?> op = executor.getWriteThenReadOperation();
+        WriteThenReadOperationCursor<?> op = executor.getWriteThenReadOperation();
         assertNotNull(op, "expected a write-then-read operation");
         assertEquals(NAMESPACE, op.getNamespace());
         // Must not fall through to the plain read-operation path.

driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/TestHelper.java

@@ -26,7 +26,7 @@
 import com.mongodb.internal.bulk.IndexRequest;
 import com.mongodb.internal.bulk.WriteRequest;
 import com.mongodb.internal.client.model.FindOptions;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.internal.operation.ReadOperation;
 import com.mongodb.internal.operation.WriteOperation;
 import com.mongodb.lang.NonNull;
@@ -97,7 +97,7 @@ public class TestHelper {
                 .execute(any(WriteOperation.class), any(), any());
         Mockito.lenient().doAnswer(invocation -> Mono.empty())
                 .when(executor)
-                .execute(any(AsyncWriteThenReadOperationCursor.class), any(), any());
+                .execute(any(WriteThenReadOperationCursor.class), any(), any());
         Mockito.lenient().doAnswer(invocation -> Mono.empty())
                 .when(executor)
                 .execute(any(), any(), any(), any());

driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/TestOperationExecutor.java

@@ -20,7 +20,7 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.internal.TimeoutSettings;
 import com.mongodb.internal.async.AsyncBatchCursor;
-import com.mongodb.internal.operation.AsyncWriteThenReadOperationCursor;
+import com.mongodb.internal.operation.WriteThenReadOperationCursor;
 import com.mongodb.internal.operation.ReadOperation;
 import com.mongodb.internal.operation.WriteOperation;
 import com.mongodb.lang.Nullable;
@@ -39,7 +39,7 @@ public class TestOperationExecutor implements OperationExecutor {
 
     private final List<ReadOperation> readOperations = new ArrayList<>();
     private final List<WriteOperation> writeOperations = new ArrayList<>();
-    private final List<AsyncWriteThenReadOperationCursor> writeThenReadOperations = new ArrayList<>();
+    private final List<WriteThenReadOperationCursor> writeThenReadOperations = new ArrayList<>();
 
     public TestOperationExecutor(final List<Object> responses) {
         this.responses = new ArrayList<>(responses);
@@ -64,7 +64,7 @@ public <T> Mono<T> execute(final WriteOperation<T> operation, final ReadConcern
     }
 
     @Override
-    public <T> Mono<AsyncBatchCursor<T>> execute(final AsyncWriteThenReadOperationCursor<T> operation, final ReadConcern readConcern,
+    public <T> Mono<AsyncBatchCursor<T>> execute(final WriteThenReadOperationCursor<T> operation, final ReadConcern readConcern,
                                                  @Nullable final ClientSession session) {
         clientSessions.add(session);
         writeThenReadOperations.add(operation);
@@ -118,7 +118,7 @@ WriteOperation getWriteOperation() {
     }
 
     @Nullable
-    AsyncWriteThenReadOperationCursor getWriteThenReadOperation() {
+    WriteThenReadOperationCursor getWriteThenReadOperation() {
         return writeThenReadOperations.isEmpty() ? null : writeThenReadOperations.remove(0);
     }