Add improved Bulk Write API for Java Reactive Driver

- Created and documented the new Reactive Bulk Write API
- Enabled unified and prose tests for reactive Bulk Write API

JAVA-5530

Author: vbabanin

driver-core/src/main/com/mongodb/internal/async/AsyncFunction.java

@@ -34,7 +34,7 @@ public interface AsyncFunction<T, R> {
      * @param value A {@code @}{@link Nullable} argument of the asynchronous function.
      * @param callback the callback
      */
-    void unsafeFinish(T value, SingleResultCallback<R> callback);
+    void unsafeFinish(@Nullable T value, SingleResultCallback<R> callback);
 
     /**
      * Must be invoked at end of async chain or when executing a callback handler supplied by the caller.

driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java

@@ -17,6 +17,8 @@
 package com.mongodb.internal.async;
 
 import com.mongodb.internal.TimeoutContext;
+import com.mongodb.internal.async.function.AsyncCallbackLoop;
+import com.mongodb.internal.async.function.LoopState;
 import com.mongodb.internal.async.function.RetryState;
 import com.mongodb.internal.async.function.RetryingAsyncCallbackSupplier;
 
@@ -181,6 +183,27 @@ default AsyncRunnable thenRun(final AsyncRunnable runnable) {
         };
     }
 
+//    /**
+//     * @param runnable The async runnable to run after this runnable
+//     * @return the composition of this runnable and the runnable, a runnable
+//     */
+//    default AsyncSupplier<T> thenSupplyUntil(
+//            final AsyncSupplier supplier,
+//            final Predicate<Boolean> condition,
+//            final Consumer<T> runnable) {
+//        return (c) -> {
+//            this.unsafeFinish((r, e) -> {
+//                if (e == null) {
+//                    /* If 'runnable' is executed on a different thread from the one that executed the initial 'finish()',
+//                     then invoking 'finish()' within 'runnable' will catch and propagate any exceptions to 'c' (the callback). */
+//                    supplier.finish(c);
+//                } else {
+//                    c.completeExceptionally(e);
+//                }
+//            });
+//        };
+//    }
+
     /**
      * The error check checks if the exception is an instance of the provided class.
      * @see #thenRunTryCatchAsyncBlocks(AsyncRunnable, java.util.function.Predicate, AsyncFunction)
@@ -217,6 +240,18 @@ default AsyncRunnable thenRunTryCatchAsyncBlocks(
         });
     }
 
+    default <R> AsyncSupplier<R> thenSupplyTryCatchAsyncBlocks(
+            final AsyncSupplier<R> supplier,
+            final Predicate<Throwable> errorCheck,
+            final AsyncFunction<Throwable, R> errorFunction) {
+        return this.thenSupply(c -> {
+            beginAsync()
+                    .thenSupply(supplier)
+                    .onErrorIf(errorCheck, errorFunction)
+                    .finish(c);
+        });
+    }
+
     /**
      * @param condition the condition to check
      * @param runnable The async runnable to run after this runnable,
@@ -282,4 +317,51 @@ default AsyncRunnable thenRunRetryingWhile(
             ).get(callback);
         });
     }
+
+    /**
+     * In order to break the loop and complete the ongoing iteration, use
+     * {@link LoopState#breakAndCompleteIf(Supplier, SingleResultCallback)} in the loopBodyRunnable.
+     *
+     * <p>
+     * This is equivalent to while(true) with break.
+     *
+     * @param loopBodyRunnable the loopBodyRunnable to loop
+     * @return the composition of this, and the looping branch
+     * @see AsyncCallbackLoop
+     */
+    default AsyncRunnable thenRunWhileLoop(final AsyncRunnable loopBodyRunnable, final LoopState loopState) {
+        return thenRun(callback -> {
+            new AsyncCallbackLoop(loopState, loopBodyRunnable::finish).run(callback);
+        });
+    }
+
+    /**
+     * This method is equivalent to a do-while loop, where the loop body is executed first and
+     * then the condition is checked to determine whether the loop should continue.
+     *
+     * @param loopBodyRunnable the asynchronous task to be executed in each iteration of the loop
+     * @param whileCheck a condition to check after each iteration; the loop continues as long as this condition returns true
+     * @return the composition of this and the looping branch
+     * @see AsyncCallbackLoop
+     */
+
+    default AsyncRunnable thenRunDoWhileLoop(final AsyncRunnable loopBodyRunnable, final Supplier<Boolean> whileCheck) {
+        return thenRun(finalCallback -> {
+            LoopState loopState = new LoopState();
+            new AsyncCallbackLoop(loopState, iterationCallback -> {
+
+                loopBodyRunnable.finish((result, t) -> {
+                    if (t != null) {
+                        iterationCallback.completeExceptionally(t);
+                        return;
+                    }
+                    if (loopState.breakAndCompleteIf(() -> !whileCheck.get(), iterationCallback)) {
+                        return;
+                    }
+                    iterationCallback.complete(iterationCallback);
+                });
+
+            }).run(finalCallback);
+        });
+    }
 }

driver-core/src/main/com/mongodb/internal/async/AsyncSupplier.java

@@ -81,6 +81,26 @@ default void finish(final SingleResultCallback<T> callback) {
         }
     }
 
+    /**
+     * The runnable will always be executed, including on the exceptional path.
+     * @param runnable the runnable
+     * @param callback the callback
+     */
+    default void thenAlwaysRunAndFinish(final Runnable runnable, final SingleResultCallback<T> callback) {
+        this.finish((r, e) -> {
+            try {
+                runnable.run();
+            } catch (Throwable t) {
+                if (e != null) {
+                    t.addSuppressed(e);
+                }
+                callback.completeExceptionally(t);
+                return;
+            }
+            callback.onResult(r, e);
+        });
+    }
+
     /**
      * @param function The async function to run after this supplier
      * @return the composition of this supplier and the function, a supplier

driver-core/src/main/com/mongodb/internal/async/MutableValue.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.internal.async;
+
+import com.mongodb.annotations.NotThreadSafe;
+import com.mongodb.assertions.Assertions;
+import com.mongodb.lang.Nullable;
+
+@NotThreadSafe
+public class MutableValue<T> {
+    private T value;
+
+    public MutableValue(@Nullable final T value) {
+        this.value = value;
+    }
+
+    public MutableValue() {
+        this(null);
+    }
+
+    public T get() {
+        return Assertions.assertNotNull(value);
+    }
+
+    @Nullable
+    public T getNullable() {
+        return value;
+    }
+
+    public void set(@Nullable final T value) {
+        this.value = value;
+    }
+}

driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackSupplier.java

@@ -15,7 +15,7 @@
  */
 package com.mongodb.internal.async.function;
 
-import com.mongodb.annotations.NotThreadSafe;
+import com.mongodb.internal.async.MutableValue;
 import com.mongodb.internal.async.SingleResultCallback;
 
 import java.util.function.Supplier;
@@ -68,16 +68,12 @@ public interface AsyncCallbackSupplier<R> {
      * This is a price we have to pay to provide a guarantee similar to that of the {@code finally} block.
      */
     default AsyncCallbackSupplier<R> whenComplete(final Runnable after) {
-        @NotThreadSafe
-        final class MutableBoolean {
-            private boolean value;
-        }
-        MutableBoolean afterExecuted = new MutableBoolean();
+        MutableValue<Boolean> afterExecuted = new MutableValue<>(false);
         Runnable trackableAfter = () -> {
             try {
                 after.run();
             } finally {
-                afterExecuted.value = true;
+                afterExecuted.set(true);
             }
         };
         return callback -> {
@@ -103,7 +99,7 @@ final class MutableBoolean {
                 primaryUnexpectedException = unexpectedException;
                 throw unexpectedException;
             } finally {
-                if (primaryUnexpectedException != null && !afterExecuted.value) {
+                if (primaryUnexpectedException != null && !afterExecuted.get()) {
                     try {
                         trackableAfter.run();
                     } catch (Throwable afterException) {

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

@@ -44,6 +44,9 @@
 import com.mongodb.client.model.SearchIndexModel;
 import com.mongodb.client.model.UpdateOptions;
 import com.mongodb.client.model.WriteModel;
+import com.mongodb.client.model.bulk.ClientBulkWriteOptions;
+import com.mongodb.client.model.bulk.ClientBulkWriteResult;
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
 import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.client.model.changestream.FullDocumentBeforeChange;
 import com.mongodb.internal.TimeoutSettings;
@@ -293,6 +296,12 @@ public AsyncWriteOperation<BulkWriteResult> bulkWrite(final List<? extends Write
         return operations.bulkWrite(requests, options);
     }
 
+    public AsyncWriteOperation<ClientBulkWriteResult> clientBulkWriteOperation(
+            final List<? extends ClientNamespacedWriteModel> clientWriteModels,
+            @Nullable final ClientBulkWriteOptions options) {
+        return operations.clientBulkWriteOperation(clientWriteModels, options);
+    }
+
     public <TResult> AsyncReadOperation<TResult> commandRead(final Bson command, final Class<TResult> resultClass) {
         return operations.commandRead(command, resultClass);
     }