JAVA-2254: Change toCollection methods to asynchronous AggregateIterable and MapReduceIterable interfaces to throw IllegalStateException instead of IllegalArgumentExce

jyemin · jyemin · commit e33ca0c3cac6 · 2016-09-22T08:45:28.000-04:00
Useful for clients who just want to write the results to a collection but do not need to iterate over them.
diff --git a/driver-async/src/main/com/mongodb/async/client/AggregateIterable.java b/driver-async/src/main/com/mongodb/async/client/AggregateIterable.java
@@ -63,6 +63,7 @@ public interface AggregateIterable<TResult> extends MongoIterable<TResult> {
      * Aggregates documents according to the specified aggregation pipeline, which must end with a $out stage.
      *
      * @param callback the callback, which is called when the aggregation completes
+     * @throws IllegalStateException if the pipeline does not end with a $out stage
      * @mongodb.driver.manual aggregation/ Aggregation
      */
     void toCollection(SingleResultCallback<Void> callback);
diff --git a/driver-async/src/main/com/mongodb/async/client/AggregateIterableImpl.java b/driver-async/src/main/com/mongodb/async/client/AggregateIterableImpl.java
@@ -107,7 +107,7 @@ public void toCollection(final SingleResultCallback<Void> callback) {
         BsonValue outCollection = getAggregateOutCollection(aggregateList);
 
         if (outCollection == null) {
-            throw new IllegalArgumentException("The last stage of the aggregation pipeline must be $out");
+            throw new IllegalStateException("The last stage of the aggregation pipeline must be $out");
         }
 
         executor.execute(new AggregateToCollectionOperation(namespace, aggregateList, writeConcern)
diff --git a/driver-async/src/main/com/mongodb/async/client/MapReduceIterable.java b/driver-async/src/main/com/mongodb/async/client/MapReduceIterable.java
@@ -190,6 +190,8 @@ public interface MapReduceIterable<TResult> extends MongoIterable<TResult> {
      * non-inline result.
      *
      * @param callback the callback, which is called when the aggregation completes
+     * @throws IllegalStateException if a collection name to write the results to has not been specified
+     * @see #collectionName(String)
      * @mongodb.driver.manual aggregation/ Aggregation
      */
     void toCollection(SingleResultCallback<Void> callback);
diff --git a/driver-async/src/main/com/mongodb/async/client/MapReduceIterableImpl.java b/driver-async/src/main/com/mongodb/async/client/MapReduceIterableImpl.java
@@ -192,7 +192,7 @@ public MapReduceIterable<TResult> collation(final Collation collation) {
     public void toCollection(final SingleResultCallback<Void> callback) {
         notNull("callback", callback);
         if (inline) {
-            throw new IllegalArgumentException("The options must specify a non-inline result");
+            throw new IllegalStateException("The options must specify a non-inline result");
         }
         executor.execute(createMapReduceToCollectionOperation(),
                 new SingleResultCallback<MapReduceStatistics>() {
diff --git a/driver-async/src/test/unit/com/mongodb/async/client/AggregateIterableSpecification.groovy b/driver-async/src/test/unit/com/mongodb/async/client/AggregateIterableSpecification.groovy
@@ -164,11 +164,11 @@ class AggregateIterableSpecification extends Specification {
         then: 'the future should handle the exception'
         thrown(MongoException)
 
-        when: 'toCollection should throw IllegalArgumentException when last state is not $out'
+        when: 'toCollection should throw IllegalStateException when last state is not $out'
         aggregationIterable.toCollection(new FutureResultCallback())
 
         then:
-        thrown(IllegalArgumentException)
+        thrown(IllegalStateException)
 
         when: 'a codec is missing'
         futureResultCallback = new FutureResultCallback<List<BsonDocument>>()
diff --git a/driver-async/src/test/unit/com/mongodb/async/client/MapReduceIterableSpecification.groovy b/driver-async/src/test/unit/com/mongodb/async/client/MapReduceIterableSpecification.groovy
@@ -193,11 +193,11 @@ class MapReduceIterableSpecification extends Specification {
         then: 'the future should handle the exception'
         thrown(MongoException)
 
-        when: 'toCollection should throw IllegalArgumentException its inline'
+        when: 'toCollection should throw IllegalStateException its inline'
         mapReduceIterable.toCollection(new FutureResultCallback())
 
         then:
-        thrown(IllegalArgumentException)
+        thrown(IllegalStateException)
 
         when: 'a codec is missing'
         futureResultCallback = new FutureResultCallback<List<BsonDocument>>()
diff --git a/driver/src/main/com/mongodb/AggregateIterableImpl.java b/driver/src/main/com/mongodb/AggregateIterableImpl.java
@@ -70,6 +70,17 @@ class AggregateIterableImpl<TDocument, TResult> implements AggregateIterable<TRe
         this.pipeline = notNull("pipeline", pipeline);
     }
 
+    @Override
+    public void toCollection() {
+        List<BsonDocument> aggregateList = createBsonDocumentList(pipeline);
+
+        if (getOutCollection(aggregateList) == null) {
+            throw new IllegalStateException("The last stage of the aggregation pipeline must be $out");
+        }
+
+        executor.execute(createAggregateToCollectionOperation(aggregateList));
+    }
+
     @Override
     public AggregateIterable<TResult> allowDiskUse(final Boolean allowDiskUse) {
         this.allowDiskUse = allowDiskUse;
@@ -135,15 +146,10 @@ public <A extends Collection<? super TResult>> A into(final A target) {
     private MongoIterable<TResult> execute() {
         List<BsonDocument> aggregateList = createBsonDocumentList(pipeline);
 
-        BsonValue outCollection = aggregateList.size() == 0 ? null : aggregateList.get(aggregateList.size() - 1).get("$out");
+        BsonValue outCollection = getOutCollection(aggregateList);
 
         if (outCollection != null) {
-            AggregateToCollectionOperation operation = new AggregateToCollectionOperation(namespace, aggregateList, writeConcern)
-                    .maxTime(maxTimeMS, MILLISECONDS)
-                    .allowDiskUse(allowDiskUse)
-                    .bypassDocumentValidation(bypassDocumentValidation)
-                    .collation(collation);
-            executor.execute(operation);
+            executor.execute(createAggregateToCollectionOperation(aggregateList));
             FindIterable<TResult> findOperation = new FindIterableImpl<TDocument, TResult>(new MongoNamespace(namespace.getDatabaseName(),
                     outCollection.asString().getValue()), documentClass, resultClass, codecRegistry, readPreference, readConcern, executor,
                     new BsonDocument(), new FindOptions().collation(collation));
@@ -163,6 +169,19 @@ private MongoIterable<TResult> execute() {
         }
     }
 
+
+    private BsonValue getOutCollection(final List<BsonDocument> aggregateList) {
+        return aggregateList.size() == 0 ? null : aggregateList.get(aggregateList.size() - 1).get("$out");
+    }
+
+    private AggregateToCollectionOperation createAggregateToCollectionOperation(final List<BsonDocument> aggregateList) {
+        return new AggregateToCollectionOperation(namespace, aggregateList, writeConcern)
+                        .maxTime(maxTimeMS, MILLISECONDS)
+                        .allowDiskUse(allowDiskUse)
+                        .bypassDocumentValidation(bypassDocumentValidation)
+                        .collation(collation);
+    }
+
     private List<BsonDocument> createBsonDocumentList(final List<? extends Bson> pipeline) {
         List<BsonDocument> aggregateList = new ArrayList<BsonDocument>(pipeline.size());
         for (Bson obj : pipeline) {
diff --git a/driver/src/main/com/mongodb/MapReduceIterableImpl.java b/driver/src/main/com/mongodb/MapReduceIterableImpl.java
@@ -83,6 +83,15 @@ class MapReduceIterableImpl<TDocument, TResult> implements MapReduceIterable<TRe
         this.reduceFunction = notNull("reduceFunction", reduceFunction);
     }
 
+    @Override
+    public void toCollection() {
+        if (inline) {
+            throw new IllegalStateException("The options must specify a non-inline result");
+        }
+
+        executor.execute(createMapReduceToCollectionOperation());
+    }
+
     @Override
     public MapReduceIterable<TResult> collectionName(final String collectionName) {
         this.collectionName = notNull("collectionName", collectionName);
@@ -227,27 +236,7 @@ MongoIterable<TResult> execute() {
             }
             return new OperationIterable<TResult>(operation, readPreference, executor);
         } else {
-            MapReduceToCollectionOperation operation =
-                    new MapReduceToCollectionOperation(namespace, new BsonJavaScript(mapFunction), new BsonJavaScript(reduceFunction),
-                            collectionName, writeConcern)
-                            .filter(toBsonDocument(filter))
-                            .limit(limit)
-                            .maxTime(maxTimeMS, MILLISECONDS)
-                            .jsMode(jsMode)
-                            .scope(toBsonDocument(scope))
-                            .sort(toBsonDocument(sort))
-                            .verbose(verbose)
-                            .action(action.getValue())
-                            .nonAtomic(nonAtomic)
-                            .sharded(sharded)
-                            .databaseName(databaseName)
-                            .bypassDocumentValidation(bypassDocumentValidation)
-                            .collation(collation);
-
-            if (finalizeFunction != null) {
-                operation.finalizeFunction(new BsonJavaScript(finalizeFunction));
-            }
-            executor.execute(operation);
+            executor.execute(createMapReduceToCollectionOperation());
 
             String dbName = databaseName != null ? databaseName : namespace.getDatabaseName();
             return new FindIterableImpl<TDocument, TResult>(new MongoNamespace(dbName, collectionName), documentClass, resultClass,
@@ -256,6 +245,30 @@ codecRegistry, primary(), readConcern, executor, new BsonDocument(),
         }
     }
 
+    private MapReduceToCollectionOperation createMapReduceToCollectionOperation() {
+        MapReduceToCollectionOperation operation =
+                new MapReduceToCollectionOperation(namespace, new BsonJavaScript(mapFunction), new BsonJavaScript(reduceFunction),
+                        collectionName, writeConcern)
+                        .filter(toBsonDocument(filter))
+                        .limit(limit)
+                        .maxTime(maxTimeMS, MILLISECONDS)
+                        .jsMode(jsMode)
+                        .scope(toBsonDocument(scope))
+                        .sort(toBsonDocument(sort))
+                        .verbose(verbose)
+                        .action(action.getValue())
+                        .nonAtomic(nonAtomic)
+                        .sharded(sharded)
+                        .databaseName(databaseName)
+                        .bypassDocumentValidation(bypassDocumentValidation)
+                        .collation(collation);
+
+        if (finalizeFunction != null) {
+            operation.finalizeFunction(new BsonJavaScript(finalizeFunction));
+        }
+        return operation;
+    }
+
     private BsonDocument toBsonDocument(final Bson document) {
         return document == null ? null : document.toBsonDocument(documentClass, codecRegistry);
     }
diff --git a/driver/src/main/com/mongodb/client/AggregateIterable.java b/driver/src/main/com/mongodb/client/AggregateIterable.java
@@ -24,10 +24,20 @@
  * Iterable for aggregate.
  *
  * @param <TResult> The type of the result.
+ * @mongodb.driver.manual reference/command/aggregate/ Aggregation
  * @since 3.0
  */
 public interface AggregateIterable<TResult> extends MongoIterable<TResult> {
 
+    /**
+     * Aggregates documents according to the specified aggregation pipeline, which must end with a $out stage.
+     *
+     * @throws IllegalStateException if the pipeline does not end with a $out stage
+     * @mongodb.driver.manual reference/operator/aggregation/out/ $out stage
+     * @since 3.4
+     */
+    void toCollection();
+
     /**
      * Enables writing to temporary files. A null value indicates that it's unspecified.
      *
diff --git a/driver/src/main/com/mongodb/client/MapReduceIterable.java b/driver/src/main/com/mongodb/client/MapReduceIterable.java
@@ -33,6 +33,16 @@
  */
 public interface MapReduceIterable<TResult> extends MongoIterable<TResult> {
 
+    /**
+     * Aggregates documents to a collection according to the specified map-reduce function with the given options, which must specify a
+     * non-inline result.
+     *
+     * @throws IllegalStateException if a collection name to write the results to has not been specified
+     * @see #collectionName(String)
+     * @since 3.4
+     */
+    void toCollection();
+
     /**
      * Sets the collectionName for the output of the MapReduce
      *
diff --git a/driver/src/test/unit/com/mongodb/AggregateIterableSpecification.groovy b/driver/src/test/unit/com/mongodb/AggregateIterableSpecification.groovy
@@ -106,6 +106,20 @@ class AggregateIterableSpecification extends Specification {
         operation.getNamespace() == collectionNamespace
         operation.getBatchSize() == 99
         operation.getCollation() == collation
+
+        when: 'toCollection should work as expected'
+        new AggregateIterableImpl(namespace, Document, Document, codecRegistry, readPreference, readConcern, writeConcern, executor,
+                pipeline)
+                .allowDiskUse(true)
+                .collation(collation)
+                .toCollection();
+
+        operation = executor.getWriteOperation() as AggregateToCollectionOperation
+
+        then:
+        expect operation, isTheSameAs(new AggregateToCollectionOperation(namespace,
+                [new BsonDocument('$match', new BsonInt32(1)), new BsonDocument('$out', new BsonString(collectionName))], writeConcern)
+                .allowDiskUse(true).collation(collation))
     }
 
     def 'should handle exceptions correctly'() {
@@ -122,6 +136,12 @@ class AggregateIterableSpecification extends Specification {
         then: 'the future should handle the exception'
         thrown(MongoException)
 
+        when: 'toCollection should throw IllegalStateException when last state is not $out'
+        aggregationIterable.toCollection()
+
+        then:
+        thrown(IllegalStateException)
+
         when: 'a codec is missing'
         new AggregateIterableImpl(namespace, Document, Document, codecRegistry, readPreference, readConcern, writeConcern, executor,
                                   pipeline).iterator()
diff --git a/driver/src/test/unit/com/mongodb/MapReduceIterableSpecification.groovy b/driver/src/test/unit/com/mongodb/MapReduceIterableSpecification.groovy
@@ -94,7 +94,7 @@ class MapReduceIterableSpecification extends Specification {
 
     def 'should build the expected MapReduceToCollectionOperation'() {
         given:
-        def executor = new TestOperationExecutor([null, null]);
+        def executor = new TestOperationExecutor([null, null, null]);
 
         when: 'mapReduce to a collection'
         def collectionNamespace = new MongoNamespace('dbName', 'collName')
@@ -146,6 +146,14 @@ class MapReduceIterableSpecification extends Specification {
         operation.getNamespace() == collectionNamespace
         operation.getBatchSize() == 99
         operation.getCollation() == collation
+
+        when: 'toCollection should work as expected'
+        mapReduceIterable.toCollection()
+
+        operation = executor.getWriteOperation() as MapReduceToCollectionOperation
+
+        then:
+        expect operation, isTheSameAs(expectedOperation)
     }
 
     def 'should handle exceptions correctly'() {
@@ -162,6 +170,12 @@ class MapReduceIterableSpecification extends Specification {
         then: 'the future should handle the exception'
         thrown(MongoException)
 
+        when: 'toCollection should throw IllegalStateException if its inline'
+        mapReduceIterable.toCollection()
+
+        then:
+        thrown(IllegalStateException)
+
         when: 'a codec is missing'
         new MapReduceIterableImpl(namespace, Document, Document, codecRegistry, readPreference, readConcern, writeConcern, executor,
                 'map', 'reduce').iterator()

Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,7 @@ public interface AggregateIterable<TResult> extends MongoIterable<TResult> {`
`63`	`63`	`* Aggregates documents according to the specified aggregation pipeline, which must end with a $out stage.`
`64`	`64`	`*`
`65`	`65`	`* @param callback the callback, which is called when the aggregation completes`
	`66`	`+ * @throws IllegalStateException if the pipeline does not end with a $out stage`
`66`	`67`	`* @mongodb.driver.manual aggregation/ Aggregation`
`67`	`68`	`*/`
`68`	`69`	`void toCollection(SingleResultCallback<Void> callback);`
Original file line number	Diff line number	Diff line change
`@@ -107,7 +107,7 @@ public void toCollection(final SingleResultCallback<Void> callback) {`
`107`	`107`	`BsonValue outCollection = getAggregateOutCollection(aggregateList);`
`108`	`108`
`109`	`109`	`if (outCollection == null) {`
`110`		`- throw new IllegalArgumentException("The last stage of the aggregation pipeline must be $out");`
	`110`	`+ throw new IllegalStateException("The last stage of the aggregation pipeline must be $out");`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`executor.execute(new AggregateToCollectionOperation(namespace, aggregateList, writeConcern)`
Original file line number	Diff line number	Diff line change
`@@ -190,6 +190,8 @@ public interface MapReduceIterable<TResult> extends MongoIterable<TResult> {`
`190`	`190`	`* non-inline result.`
`191`	`191`	`*`
`192`	`192`	`* @param callback the callback, which is called when the aggregation completes`
	`193`	`+ * @throws IllegalStateException if a collection name to write the results to has not been specified`
	`194`	`+ * @see #collectionName(String)`
`193`	`195`	`* @mongodb.driver.manual aggregation/ Aggregation`
`194`	`196`	`*/`
`195`	`197`	`void toCollection(SingleResultCallback<Void> callback);`
Original file line number	Diff line number	Diff line change
`@@ -192,7 +192,7 @@ public MapReduceIterable<TResult> collation(final Collation collation) {`
`192`	`192`	`public void toCollection(final SingleResultCallback<Void> callback) {`
`193`	`193`	`notNull("callback", callback);`
`194`	`194`	`if (inline) {`
`195`		`- throw new IllegalArgumentException("The options must specify a non-inline result");`
	`195`	`+ throw new IllegalStateException("The options must specify a non-inline result");`
`196`	`196`	`}`
`197`	`197`	`executor.execute(createMapReduceToCollectionOperation(),`
`198`	`198`	`new SingleResultCallback<MapReduceStatistics>() {`