Improve the toCollection and related API documentation (mongodb#1677)

JAVA-5833

Author: stIncMale

driver-kotlin-coroutine/src/main/kotlin/com/mongodb/kotlin/client/coroutine/AggregateFlow.kt

@@ -62,9 +62,11 @@ public class AggregateFlow<T : Any>(private val wrapped: AggregatePublisher<T>)
     public fun timeoutMode(timeoutMode: TimeoutMode): AggregateFlow<T> = apply { wrapped.timeoutMode(timeoutMode) }
 
     /**
-     * Aggregates documents according to the specified aggregation pipeline, which must end with a $out or $merge stage.
+     * Aggregates documents according to the specified aggregation pipeline, which must end with an `$out` or `$merge`
+     * stage. Calling this method is the preferred alternative to consuming this [AggregateFlow], because this method
+     * does what is explicitly requested without executing implicit operations.
      *
-     * @throws IllegalStateException if the pipeline does not end with a $out or $merge stage
+     * @throws IllegalStateException if the pipeline does not end with an `$out` or `$merge` stage
      * @see [$out stage](https://www.mongodb.com/docs/manual/reference/operator/aggregation/out/)
      * @see [$merge stage](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge/)
      */
@@ -214,5 +216,11 @@ public class AggregateFlow<T : Any>(private val wrapped: AggregatePublisher<T>)
     public suspend inline fun <reified R : Any> explain(verbosity: ExplainVerbosity? = null): R =
         explain(R::class.java, verbosity)
 
+    /**
+     * Requests [AggregateFlow] to start streaming data according to the specified aggregation pipeline.
+     * - If the aggregation pipeline ends with an `$out` or `$merge` stage, then finds all documents in the affected
+     *   namespace and emits them. You may want to use [toCollection] instead.
+     * - Otherwise, emits no values.
+     */
     public override suspend fun collect(collector: FlowCollector<T>): Unit = wrapped.asFlow().collect(collector)
 }

driver-kotlin-coroutine/src/main/kotlin/com/mongodb/kotlin/client/coroutine/MapReduceFlow.kt

@@ -33,6 +33,9 @@ import org.bson.conversions.Bson
 /**
  * Flow implementation for map reduce operations.
  *
+ * By default, the [MapReduceFlow] emits the results inline. You can write map-reduce output to a collection by using
+ * the [collectionName] and [toCollection] methods.
+ *
  * Note: Starting in MongoDB 5.0, map-reduce is deprecated, prefer Aggregation instead
  *
  * @param T The type of the result.
@@ -65,9 +68,11 @@ public class MapReduceFlow<T : Any>(private val wrapped: MapReducePublisher<T>)
 
     /**
      * Aggregates documents to a collection according to the specified map-reduce function with the given options, which
-     * must specify a non-inline result.
+     * must not emit results inline. Calling this method is the preferred alternative to consuming this [MapReduceFlow],
+     * because this method does what is explicitly requested without executing implicit operations.
      *
      * @throws IllegalStateException if a collection name to write the results to has not been specified
+     * @see collectionName
      */
     public suspend fun toCollection() {
         wrapped.toCollection().awaitFirstOrNull()
@@ -80,6 +85,7 @@ public class MapReduceFlow<T : Any>(private val wrapped: MapReducePublisher<T>)
      *
      * @param collectionName the name of the collection that you want the map-reduce operation to write its output.
      * @return this
+     * @see toCollection
      */
     public fun collectionName(collectionName: String): MapReduceFlow<T> = apply {
         wrapped.collectionName(collectionName)
@@ -205,5 +211,12 @@ public class MapReduceFlow<T : Any>(private val wrapped: MapReducePublisher<T>)
      */
     public fun collation(collation: Collation?): MapReduceFlow<T> = apply { wrapped.collation(collation) }
 
+    /**
+     * Requests [MapReduceFlow] to start streaming data according to the specified map-reduce function with the given
+     * options.
+     * - If the aggregation produces results inline, then finds all documents in the affected namespace and emits them.
+     *   You may want to use [toCollection] instead.
+     * - Otherwise, emits no values.
+     */
     public override suspend fun collect(collector: FlowCollector<T>): Unit = wrapped.asFlow().collect(collector)
 }

driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/AggregateFlowTest.kt

@@ -39,7 +39,8 @@ class AggregateFlowTest {
 
     @Test
     fun shouldHaveTheSameMethods() {
-        val jAggregatePublisherFunctions = AggregatePublisher::class.declaredFunctions.map { it.name }.toSet() - "first"
+        val jAggregatePublisherFunctions =
+            AggregatePublisher::class.declaredFunctions.map { it.name }.toSet() - "first" - "subscribe"
         val kAggregateFlowFunctions = AggregateFlow::class.declaredFunctions.map { it.name }.toSet() - "collect"
 
         assertEquals(jAggregatePublisherFunctions, kAggregateFlowFunctions)

driver-kotlin-coroutine/src/test/kotlin/com/mongodb/kotlin/client/coroutine/MapReduceFlowTest.kt

@@ -38,7 +38,8 @@ import reactor.core.publisher.Mono
 class MapReduceFlowTest {
     @Test
     fun shouldHaveTheSameMethods() {
-        val jMapReducePublisherFunctions = MapReducePublisher::class.declaredFunctions.map { it.name }.toSet() - "first"
+        val jMapReducePublisherFunctions =
+            MapReducePublisher::class.declaredFunctions.map { it.name }.toSet() - "first" - "subscribe"
         val kMapReduceFlowFunctions = MapReduceFlow::class.declaredFunctions.map { it.name }.toSet() - "collect"
 
         assertEquals(jMapReducePublisherFunctions, kMapReduceFlowFunctions)

driver-kotlin-sync/src/main/kotlin/com/mongodb/kotlin/client/AggregateIterable.kt

@@ -61,14 +61,24 @@ public class AggregateIterable<T : Any>(private val wrapped: JAggregateIterable<
     }
 
     /**
-     * Aggregates documents according to the specified aggregation pipeline, which must end with a $out or $merge stage.
+     * Aggregates documents according to the specified aggregation pipeline, which must end with an `$out` or `$merge`
+     * stage. This method is the preferred alternative to [cursor], because this method does what is explicitly
+     * requested without executing implicit operations.
      *
-     * @throws IllegalStateException if the pipeline does not end with a $out or $merge stage
+     * @throws IllegalStateException if the pipeline does not end with an `$out` or `$merge` stage
      * @see [$out stage](https://www.mongodb.com/docs/manual/reference/operator/aggregation/out/)
      * @see [$merge stage](https://www.mongodb.com/docs/manual/reference/operator/aggregation/merge/)
      */
     public fun toCollection(): Unit = wrapped.toCollection()
 
+    /**
+     * Aggregates documents according to the specified aggregation pipeline.
+     * - If the aggregation pipeline ends with an `$out` or `$merge` stage, then finds all documents in the affected
+     *   namespace and returns a [MongoCursor] over them. You may want to use [toCollection] instead.
+     * - Otherwise, returns a [MongoCursor] producing no elements.
+     */
+    public override fun cursor(): MongoCursor<T> = super.cursor()
+
     /**
      * Enables writing to temporary files. A null value indicates that it's unspecified.
      *

driver-kotlin-sync/src/test/kotlin/com/mongodb/kotlin/client/AggregateIterableTest.kt

@@ -37,7 +37,8 @@ class AggregateIterableTest {
 
     @Test
     fun shouldHaveTheSameMethods() {
-        val jAggregateIterableFunctions = JAggregateIterable::class.declaredFunctions.map { it.name }.toSet()
+        val jAggregateIterableFunctions =
+            JAggregateIterable::class.declaredFunctions.map { it.name }.toSet() - "iterator"
         val kAggregateIterableFunctions = AggregateIterable::class.declaredFunctions.map { it.name }.toSet()
 
         assertEquals(jAggregateIterableFunctions, kAggregateIterableFunctions)

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

@@ -17,15 +17,19 @@
 package com.mongodb.reactivestreams.client;
 
 import com.mongodb.ExplainVerbosity;
+import com.mongodb.MongoNamespace;
 import com.mongodb.annotations.Alpha;
 import com.mongodb.annotations.Reason;
 import com.mongodb.client.cursor.TimeoutMode;
+import com.mongodb.client.model.Aggregates;
 import com.mongodb.client.model.Collation;
+import com.mongodb.client.model.MergeOptions;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonValue;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 import java.util.concurrent.TimeUnit;
 
@@ -83,13 +87,32 @@ public interface AggregatePublisher<TResult> extends Publisher<TResult> {
     AggregatePublisher<TResult> bypassDocumentValidation(@Nullable Boolean bypassDocumentValidation);
 
     /**
-     * Aggregates documents according to the specified aggregation pipeline, which must end with a $out stage.
+     * Aggregates documents according to the specified aggregation pipeline, which must end with an
+     * {@link Aggregates#out(String, String) $out} or {@link Aggregates#merge(MongoNamespace, MergeOptions) $merge} stage.
+     * Calling this method and then {@linkplain Publisher#subscribe(Subscriber) subscribing} to the returned {@link Publisher}
+     * is the preferred alternative to {@linkplain #subscribe(Subscriber) subscribing} to this {@link AggregatePublisher},
+     * because this method does what is explicitly requested without executing implicit operations.
      *
+     * @throws IllegalStateException if the pipeline does not end with an {@code $out} or {@code $merge} stage
      * @return an empty publisher that indicates when the operation has completed
      * @mongodb.driver.manual aggregation/ Aggregation
      */
     Publisher<Void> toCollection();
 
+    /**
+     * Requests {@link AggregatePublisher} to start streaming data according to the specified aggregation pipeline.
+     * <ul>
+     *     <li>
+     *     If the aggregation pipeline ends with an {@link Aggregates#out(String, String) $out} or
+     *     {@link Aggregates#merge(MongoNamespace, MergeOptions) $merge} stage,
+     *     then {@linkplain MongoCollection#find() finds all} documents in the affected namespace and produces them.
+     *     You may want to use {@link #toCollection()} instead.</li>
+     *     <li>
+     *     Otherwise, produces no elements.</li>
+     * </ul>
+     */
+    void subscribe(Subscriber<? super TResult> s);
+
     /**
      * Sets the collation options
      *

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

@@ -16,19 +16,22 @@
 
 package com.mongodb.reactivestreams.client;
 
-
 import com.mongodb.annotations.Alpha;
 import com.mongodb.annotations.Reason;
 import com.mongodb.client.cursor.TimeoutMode;
 import com.mongodb.client.model.Collation;
 import com.mongodb.lang.Nullable;
 import org.bson.conversions.Bson;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 
 import java.util.concurrent.TimeUnit;
 
 /**
  * Publisher for map reduce.
+ * <p>
+ * By default, the {@code MapReducePublisher} produces the results inline. You can write map-reduce output to a collection by using the
+ * {@link #collectionName(String)} and {@link #toCollection()} methods.</p>
  *
  * @param <TResult> The type of the result.
  * @since 1.0
@@ -44,6 +47,7 @@ public interface MapReducePublisher<TResult> extends Publisher<TResult> {
      *
      * @param collectionName the name of the collection that you want the map-reduce operation to write its output.
      * @return this
+     * @see #toCollection()
      */
     MapReducePublisher<TResult> collectionName(String collectionName);
 
@@ -152,14 +156,30 @@ public interface MapReducePublisher<TResult> extends Publisher<TResult> {
     MapReducePublisher<TResult> bypassDocumentValidation(@Nullable Boolean bypassDocumentValidation);
 
     /**
-     * Aggregates documents to a collection according to the specified map-reduce function with the given options, which must specify a
-     * non-inline result.
+     * Aggregates documents to a collection according to the specified map-reduce function with the given options, which must not produce
+     * results inline. Calling this method and then {@linkplain Publisher#subscribe(Subscriber) subscribing} to the returned
+     * {@link Publisher} is the preferred alternative to {@linkplain #subscribe(Subscriber) subscribing} to this {@link MapReducePublisher},
+     * because this method does what is explicitly requested without executing implicit operations.
      *
      * @return an empty publisher that indicates when the operation has completed
+     * @throws IllegalStateException if a {@linkplain #collectionName(String) collection name} to write the results to has not been specified
+     * @see #collectionName(String)
      * @mongodb.driver.manual aggregation/ Aggregation
      */
     Publisher<Void> toCollection();
 
+    /**
+     * Requests {@link MapReducePublisher} to start streaming data according to the specified map-reduce function with the given options.
+     * <ul>
+     *     <li>
+     *     If the aggregation produces results inline, then {@linkplain MongoCollection#find() finds all} documents in the
+     *     affected namespace and produces them. You may want to use {@link #toCollection()} instead.</li>
+     *     <li>
+     *     Otherwise, produces no elements.</li>
+     * </ul>
+     */
+    void subscribe(Subscriber<? super TResult> s);
+
     /**
      * Sets the collation options
      *

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

@@ -25,6 +25,7 @@ import org.mongodb.scala.bson.BsonValue
 import org.mongodb.scala.bson.DefaultHelper.DefaultsTo
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.model.Collation
+import org.reactivestreams.Subscriber
 
 import scala.concurrent.duration.Duration
 import scala.reflect.ClassTag
@@ -192,9 +193,14 @@ case class AggregateObservable[TResult](private val wrapped: AggregatePublisher[
   }
 
   /**
-   * Aggregates documents according to the specified aggregation pipeline, which must end with a `\$out` stage.
+   * Aggregates documents according to the specified aggregation pipeline, which must end with an `\$out` or `\$merge` stage.
+   * Calling this method and then `subscribing` to the returned [[SingleObservable]]
+   * is the preferred alternative to subscribing to this [[AggregateObservable]],
+   * because this method does what is explicitly requested without executing implicit operations.
    *
    * [[https://www.mongodb.com/docs/manual/aggregation/ Aggregation]]
+   *
+   * @throws java.lang.IllegalStateException if the pipeline does not end with an `\$out` or `\$merge` stage
    * @return an Observable that indicates when the operation has completed.
    */
   def toCollection(): SingleObservable[Unit] = wrapped.toCollection()
@@ -257,5 +263,23 @@ case class AggregateObservable[TResult](private val wrapped: AggregatePublisher[
   )(implicit e: ExplainResult DefaultsTo Document, ct: ClassTag[ExplainResult]): SingleObservable[ExplainResult] =
     wrapped.explain[ExplainResult](ct, verbosity)
 
+  /**
+   * Requests [[AggregateObservable]] to start streaming data according to the specified aggregation pipeline.
+   *
+   *  - If the aggregation pipeline ends with an `\$out` or `\$merge` stage,
+   *    then finds all documents in the affected namespace and produces them.
+   *    You may want to use [[toCollection]] instead.
+   *  - Otherwise, produces no elements.
+   */
   override def subscribe(observer: Observer[_ >: TResult]): Unit = wrapped.subscribe(observer)
+
+  /**
+   * Requests [[AggregateObservable]] to start streaming data according to the specified aggregation pipeline.
+   *
+   *  - If the aggregation pipeline ends with an `\$out` or `\$merge` stage,
+   *    then finds all documents in the affected namespace and produces them.
+   *    You may want to use [[toCollection]] instead.
+   *  - Otherwise, produces no elements.
+   */
+  override def subscribe(observer: Subscriber[_ >: TResult]): Unit = wrapped.subscribe(observer)
 }

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

@@ -23,12 +23,16 @@ import com.mongodb.client.model.MapReduceAction
 import com.mongodb.reactivestreams.client.MapReducePublisher
 import org.mongodb.scala.bson.conversions.Bson
 import org.mongodb.scala.model.Collation
+import org.reactivestreams.Subscriber
 
 import scala.concurrent.duration.Duration
 
 /**
  * Observable for map reduce.
  *
+ * By default, the [[MapReduceObservable]] produces the results inline. You can write map-reduce output to a collection by using the
+ * [[collectionName]] and [[toCollection]] methods.
+ *
  * @define docsRef https://www.mongodb.com/docs/manual/reference
  *
  * @tparam TResult The type of the result.
@@ -44,6 +48,7 @@ case class MapReduceObservable[TResult](wrapped: MapReducePublisher[TResult]) ex
    *
    * @param collectionName the name of the collection that you want the map-reduce operation to write its output.
    * @return this
+   * @see [[toCollection]]
    */
   def collectionName(collectionName: String): MapReduceObservable[TResult] = {
     wrapped.collectionName(collectionName)
@@ -214,11 +219,15 @@ case class MapReduceObservable[TResult](wrapped: MapReducePublisher[TResult]) ex
   }
 
   /**
-   * Aggregates documents to a collection according to the specified map-reduce function with the given options, which must specify a
-   * non-inline result.
+   * Aggregates documents to a collection according to the specified map-reduce function with the given options, which must not produce
+   * results inline. Calling this method and then subscribing to the returned [[SingleObservable]] is the preferred alternative to
+   * subscribing to this [[MapReduceObservable]],
+   * because this method does what is explicitly requested without executing implicit operations.
    *
    * @return an Observable that indicates when the operation has completed
    * [[https://www.mongodb.com/docs/manual/aggregation/ Aggregation]]
+   * @throws java.lang.IllegalStateException if a collection name to write the results to has not been specified
+   * @see [[collectionName]]
    */
   def toCollection(): SingleObservable[Unit] = wrapped.toCollection()
 
@@ -246,5 +255,21 @@ case class MapReduceObservable[TResult](wrapped: MapReducePublisher[TResult]) ex
    */
   def first(): SingleObservable[TResult] = wrapped.first()
 
+  /**
+   * Requests [[MapReduceObservable]] to start streaming data according to the specified map-reduce function with the given options.
+   *
+   *  - If the aggregation produces results inline, then finds all documents in the
+   *    affected namespace and produces them. You may want to use [[toCollection]] instead.
+   *  - Otherwise, produces no elements.
+   */
   override def subscribe(observer: Observer[_ >: TResult]): Unit = wrapped.subscribe(observer)
+
+  /**
+   * Requests [[MapReduceObservable]] to start streaming data according to the specified map-reduce function with the given options.
+   *
+   *  - If the aggregation produces results inline, then finds all documents in the
+   *    affected namespace and produces them. You may want to use [[toCollection]] instead.
+   *  - Otherwise, produces no elements.
+   */
+  override def subscribe(observer: Subscriber[_ >: TResult]): Unit = wrapped.subscribe(observer)
 }