Add Scala Client Bulk Write API. (#1603)

JAVA-5531

Author: vbabanin

driver-scala/src/integration/scala/org/mongodb/scala/syncadapter/SyncMongoCluster.scala

@@ -1,6 +1,5 @@
 package org.mongodb.scala.syncadapter
 
-import com.mongodb.assertions.Assertions
 import com.mongodb.client.model.bulk.{ ClientBulkWriteOptions, ClientBulkWriteResult, ClientNamespacedWriteModel }
 import com.mongodb.{ ClientSessionOptions, ReadConcern, ReadPreference, WriteConcern }
 import com.mongodb.client.{ ClientSession, MongoCluster => JMongoCluster, MongoDatabase => JMongoDatabase }
@@ -129,33 +128,21 @@ class SyncMongoCluster(wrapped: MongoCluster) extends JMongoCluster {
 
   override def bulkWrite(
       models: util.List[_ <: ClientNamespacedWriteModel]
-  ): ClientBulkWriteResult = {
-    org.junit.Assume.assumeTrue("TODO-JAVA-5531 implement", java.lang.Boolean.parseBoolean(toString))
-    throw Assertions.fail("TODO-JAVA-5531 implement")
-  }
+  ): ClientBulkWriteResult = wrapped.bulkWrite(models.asScala.toList).toFuture().get()
 
   override def bulkWrite(
       models: util.List[_ <: ClientNamespacedWriteModel],
       options: ClientBulkWriteOptions
-  ): ClientBulkWriteResult = {
-    org.junit.Assume.assumeTrue("TODO-JAVA-5531 implement", java.lang.Boolean.parseBoolean(toString))
-    throw Assertions.fail("TODO-JAVA-5531 implement")
-  }
+  ): ClientBulkWriteResult = wrapped.bulkWrite(models.asScala.toList, options).toFuture().get()
 
   override def bulkWrite(
       clientSession: ClientSession,
       models: util.List[_ <: ClientNamespacedWriteModel]
-  ): ClientBulkWriteResult = {
-    org.junit.Assume.assumeTrue("TODO-JAVA-5531 implement", java.lang.Boolean.parseBoolean(toString))
-    throw Assertions.fail("TODO-JAVA-5531 implement")
-  }
+  ): ClientBulkWriteResult = wrapped.bulkWrite(unwrap(clientSession), models.asScala.toList).toFuture().get()
 
   override def bulkWrite(
       clientSession: ClientSession,
       models: util.List[_ <: ClientNamespacedWriteModel],
       options: ClientBulkWriteOptions
-  ): ClientBulkWriteResult = {
-    org.junit.Assume.assumeTrue("TODO-JAVA-5531 implement", java.lang.Boolean.parseBoolean(toString))
-    throw Assertions.fail("TODO-JAVA-5531 implement")
-  }
+  ): ClientBulkWriteResult = wrapped.bulkWrite(unwrap(clientSession), models.asScala.toList, options).toFuture().get()
 }

driver-scala/src/main/scala/org/mongodb/scala/MongoCluster.scala

@@ -22,6 +22,9 @@ import com.mongodb.reactivestreams.client.{ MongoCluster => JMongoCluster }
 import org.bson.codecs.configuration.CodecRegistry
 import org.mongodb.scala.bson.DefaultHelper.DefaultsTo
 import org.mongodb.scala.bson.conversions.Bson
+import org.mongodb.scala.model.bulk.ClientNamespacedUpdateManyModel
+import org.mongodb.scala.model.bulk.ClientNamespacedDeleteManyModel
+import org.mongodb.scala.model.bulk.{ ClientBulkWriteOptions, ClientBulkWriteResult, ClientNamespacedWriteModel }
 
 import scala.collection.JavaConverters._
 import scala.concurrent.duration.{ Duration, MILLISECONDS }
@@ -290,4 +293,124 @@ class MongoCluster(private val wrapped: JMongoCluster) {
   )(implicit e: C DefaultsTo Document, ct: ClassTag[C]): ChangeStreamObservable[C] =
     ChangeStreamObservable(wrapped.watch(clientSession, pipeline.asJava, ct))
 
+  /**
+   * Executes a client-level bulk write operation.
+   * This method is functionally equivalent to `bulkWrite(List, ClientBulkWriteOptions)`
+   * with the [[ClientBulkWriteOptions.clientBulkWriteOptions default options]].
+   *
+   * This operation supports retryable writes.
+   * Depending on the number of `models`, encoded size of `models`, and the size limits in effect,
+   * executing this operation may require multiple `bulkWrite` commands.
+   * The eligibility for retries is determined per each `bulkWrite` command:
+   * [[ClientNamespacedUpdateManyModel]], [[ClientNamespacedDeleteManyModel]] in a command render it non-retryable.
+   *
+   * This operation is not supported by MongoDB Atlas Serverless instances.
+   *
+   * [[https://www.mongodb.com/docs/manual/reference/command/bulkWrite/ bulkWrite]]
+   * @param models The [[ClientNamespacedWriteModel]] individual write operations.
+   * @return The [[SingleObservable]] signalling at most one element [[ClientBulkWriteResult]] if the operation is successful,
+   *         or the following errors:
+   *         - [[ClientBulkWriteException]]: If and only if the operation is unsuccessful or partially unsuccessful,
+   *           and there is at least one of the following pieces of information to report:
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteConcernErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getPartialResult]].
+   *         - [[MongoException]]: Only if the operation is unsuccessful.
+   * @since 5.4
+   * @note Requires MongoDB 8.0 or greater.
+   */
+  def bulkWrite(models: List[_ <: ClientNamespacedWriteModel]): SingleObservable[ClientBulkWriteResult] =
+    wrapped.bulkWrite(models.asJava)
+
+  /**
+   * Executes a client-level bulk write operation.
+   *
+   * This operation supports retryable writes.
+   * Depending on the number of `models`, encoded size of `models`, and the size limits in effect,
+   * executing this operation may require multiple `bulkWrite` commands.
+   * The eligibility for retries is determined per each `bulkWrite` command:
+   * [[ClientNamespacedUpdateManyModel]], [[ClientNamespacedDeleteManyModel]] in a command render it non-retryable.
+   *
+   * This operation is not supported by MongoDB Atlas Serverless instances.
+   *
+   * [[https://www.mongodb.com/docs/manual/reference/command/bulkWrite/ bulkWrite]]
+   * @param models The [[ClientNamespacedWriteModel]] individual write operations.
+   * @param options The options.
+   * @return The [[SingleObservable]] signalling at most one element [[ClientBulkWriteResult]] if the operation is successful,
+   *         or the following errors:
+   *         - [[ClientBulkWriteException]]: If and only if the operation is unsuccessful or partially unsuccessful,
+   *           and there is at least one of the following pieces of information to report:
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteConcernErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getPartialResult]].
+   *         - [[MongoException]]: Only if the operation is unsuccessful.
+   * @since 5.4
+   * @note Requires MongoDB 8.0 or greater.
+   */
+  def bulkWrite(
+      models: List[_ <: ClientNamespacedWriteModel],
+      options: ClientBulkWriteOptions
+  ): SingleObservable[ClientBulkWriteResult] = wrapped.bulkWrite(models.asJava, options)
+
+  /**
+   * Executes a client-level bulk write operation.
+   * This method is functionally equivalent to `bulkWrite(ClientSession, List, ClientBulkWriteOptions)`
+   * with the [[ClientBulkWriteOptions.clientBulkWriteOptions default options]].
+   *
+   * This operation supports retryable writes.
+   * Depending on the number of `models`, encoded size of `models`, and the size limits in effect,
+   * executing this operation may require multiple `bulkWrite` commands.
+   * The eligibility for retries is determined per each `bulkWrite` command:
+   * [[ClientNamespacedUpdateManyModel]], [[ClientNamespacedDeleteManyModel]] in a command render it non-retryable.
+   *
+   * This operation is not supported by MongoDB Atlas Serverless instances.
+   *
+   * [[https://www.mongodb.com/docs/manual/reference/command/bulkWrite/ bulkWrite]]
+   * @param clientSession [[ClientSession client session]] with which to associate this operation.
+   * @param models The [[ClientNamespacedWriteModel]] individual write operations.
+   * @return The [[SingleObservable]] signalling at most one element [[ClientBulkWriteResult]] if the operation is successful,
+   *         or the following errors:
+   *         - [[ClientBulkWriteException]]: If and only if the operation is unsuccessful or partially unsuccessful,
+   *           and there is at least one of the following pieces of information to report:
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteConcernErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getPartialResult]].
+   *         - [[MongoException]]: Only if the operation is unsuccessful.
+   * @since 5.4
+   * @note Requires MongoDB 8.0 or greater.
+   */
+  def bulkWrite(
+      clientSession: ClientSession,
+      models: List[_ <: ClientNamespacedWriteModel]
+  ): SingleObservable[ClientBulkWriteResult] = wrapped.bulkWrite(clientSession, models.asJava)
+
+  /**
+   * Executes a client-level bulk write operation.
+   *
+   * This operation supports retryable writes.
+   * Depending on the number of `models`, encoded size of `models`, and the size limits in effect,
+   * executing this operation may require multiple `bulkWrite` commands.
+   * The eligibility for retries is determined per each `bulkWrite` command:
+   * [[ClientNamespacedUpdateManyModel]], [[ClientNamespacedDeleteManyModel]] in a command render it non-retryable.
+   *
+   * [[https://www.mongodb.com/docs/manual/reference/command/bulkWrite/ bulkWrite]]
+   * @param clientSession The [[ClientSession client session]] with which to associate this operation.
+   * @param models The [[ClientNamespacedWriteModel]] individual write operations.
+   * @param options The options.
+   * @return The [[SingleObservable]] signalling at most one element [[ClientBulkWriteResult]] if the operation is successful,
+   *         or the following errors:
+   *         - [[ClientBulkWriteException]]: If and only if the operation is unsuccessful or partially unsuccessful,
+   *           and there is at least one of the following pieces of information to report:
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteConcernErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getWriteErrors]],
+   *           [[ClientBulkWriteException ClientBulkWriteException#getPartialResult]].
+   *         - [[MongoException]]: Only if the operation is unsuccessful.
+   * @since 5.4
+   * @note Requires MongoDB 8.0 or greater.
+   */
+  def bulkWrite(
+      clientSession: ClientSession,
+      models: List[_ <: ClientNamespacedWriteModel],
+      options: ClientBulkWriteOptions
+  ): SingleObservable[ClientBulkWriteResult] = wrapped.bulkWrite(clientSession, models.asJava, options)
 }