diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperation.java index f2bc62492..c872358bb 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperation.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperation.java @@ -15,9 +15,10 @@ */ package org.springframework.data.mongodb.core; +import java.util.stream.Stream; + import org.springframework.data.mongodb.core.aggregation.Aggregation; import org.springframework.data.mongodb.core.aggregation.AggregationResults; -import org.springframework.data.util.CloseableIterator; /** * {@link ExecutableAggregationOperation} allows creation and execution of MongoDB aggregation operations in a fluent @@ -88,12 +89,12 @@ public interface ExecutableAggregationOperation { /** * Apply pipeline operations as specified and stream all matching elements.
- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.FindIterable} + * Returns a {@link Stream} that wraps the Mongo DB {@link com.mongodb.client.FindIterable} * - * @return a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.FindIterable} that needs to be closed. - * Never {@literal null}. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). */ - CloseableIterator stream(); + Stream stream(); } /** diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperationSupport.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperationSupport.java index 9ac2497d0..cb9e5efa2 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperationSupport.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableAggregationOperationSupport.java @@ -15,10 +15,11 @@ */ package org.springframework.data.mongodb.core; +import java.util.stream.Stream; + import org.springframework.data.mongodb.core.aggregation.Aggregation; import org.springframework.data.mongodb.core.aggregation.AggregationResults; import org.springframework.data.mongodb.core.aggregation.TypedAggregation; -import org.springframework.data.util.CloseableIterator; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -87,7 +88,7 @@ class ExecutableAggregationOperationSupport implements ExecutableAggregationOper } @Override - public CloseableIterator stream() { + public Stream stream() { return template.aggregateStream(aggregation, getCollectionName(aggregation), domainType); } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java index f41af5c6c..5dab55db3 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java @@ -118,8 +118,8 @@ public interface ExecutableFindOperation { /** * Stream all matching elements. * - * @return a {@link Stream} that wraps the a Mongo DB {@link com.mongodb.client.FindIterable} that needs to be closed. Never - * {@literal null}. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). */ Stream stream(); diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java index 127ddb8aa..5e8cd026b 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java @@ -20,12 +20,11 @@ import java.util.Optional; import java.util.stream.Stream; import org.bson.Document; + import org.springframework.dao.IncorrectResultSizeDataAccessException; import org.springframework.data.mongodb.core.query.NearQuery; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.SerializationUtils; -import org.springframework.data.util.CloseableIterator; -import org.springframework.data.util.StreamUtils; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -70,11 +69,11 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation { private final MongoTemplate template; private final Class domainType; private final Class returnType; - @Nullable private final String collection; + private final @Nullable String collection; private final Query query; ExecutableFindSupport(MongoTemplate template, Class domainType, Class returnType, - String collection, Query query) { + @Nullable String collection, Query query) { this.template = template; this.domainType = domainType; this.returnType = returnType; @@ -137,7 +136,7 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation { @Override public Stream stream() { - return StreamUtils.createStreamFromIterator(doStream()); + return doStream(); } @Override @@ -179,7 +178,7 @@ class ExecutableFindOperationSupport implements ExecutableFindOperation { returnType == domainType ? (Class) Object.class : returnType); } - private CloseableIterator doStream() { + private Stream doStream() { return template.doStream(query, domainType, getCollectionName(), returnType); } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java index c015fb5a4..24a3223de 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2021 the original author or authors. + * Copyright 2011-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,8 +20,10 @@ import java.util.List; import java.util.Set; import java.util.function.Consumer; import java.util.function.Supplier; +import java.util.stream.Stream; import org.bson.Document; + import org.springframework.data.geo.GeoResults; import org.springframework.data.mongodb.core.BulkOperations.BulkMode; import org.springframework.data.mongodb.core.aggregation.Aggregation; @@ -42,7 +44,6 @@ import org.springframework.data.mongodb.core.query.NearQuery; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import org.springframework.data.mongodb.core.query.UpdateDefinition; -import org.springframework.data.util.CloseableIterator; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -225,34 +226,34 @@ public interface MongoOperations extends FluentMongoOperations { * Executes the given {@link Query} on the entity collection of the specified {@code entityType} backed by a Mongo DB * {@link com.mongodb.client.FindIterable}. *

- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.FindIterable} that needs to - * be closed. + * Returns a {@link String} that wraps the Mongo DB {@link com.mongodb.client.FindIterable} that needs to be closed. * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification. Must not be {@literal null}. * @param entityType must not be {@literal null}. * @param element return type - * @return will never be {@literal null}. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). * @since 1.7 */ - CloseableIterator stream(Query query, Class entityType); + Stream stream(Query query, Class entityType); /** * Executes the given {@link Query} on the entity collection of the specified {@code entityType} and collection backed * by a Mongo DB {@link com.mongodb.client.FindIterable}. *

- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.FindIterable} that needs to - * be closed. + * Returns a {@link Stream} that wraps the Mongo DB {@link com.mongodb.client.FindIterable} that needs to be closed. * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification. Must not be {@literal null}. * @param entityType must not be {@literal null}. * @param collectionName must not be {@literal null} or empty. * @param element return type - * @return will never be {@literal null}. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). * @since 1.10 */ - CloseableIterator stream(Query query, Class entityType, String collectionName); + Stream stream(Query query, Class entityType, String collectionName); /** * Create an uncapped collection with a name based on the provided entity class. @@ -521,9 +522,9 @@ public interface MongoOperations extends FluentMongoOperations { /** * Execute an aggregation operation backed by a Mongo DB {@link com.mongodb.client.AggregateIterable}. *

- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.AggregateIterable} that - * needs to be closed. The raw results will be mapped to the given entity class. The name of the inputCollection is - * derived from the inputType of the aggregation. + * Returns a {@link Stream} that wraps the Mongo DB {@link com.mongodb.client.AggregateIterable} that needs to be + * closed. The raw results will be mapped to the given entity class. The name of the inputCollection is derived from + * the inputType of the aggregation. *

* Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling * explanation mode will throw an {@link IllegalArgumentException}. @@ -532,35 +533,37 @@ public interface MongoOperations extends FluentMongoOperations { * {@literal null}. * @param collectionName The name of the input collection to use for the aggreation. * @param outputType The parametrized type of the returned list, must not be {@literal null}. - * @return The results of the aggregation operation. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). * @since 2.0 */ - CloseableIterator aggregateStream(TypedAggregation aggregation, String collectionName, Class outputType); + Stream aggregateStream(TypedAggregation aggregation, String collectionName, Class outputType); /** * Execute an aggregation operation backed by a Mongo DB {@link com.mongodb.client.AggregateIterable}. - *
- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.AggregateIterable} that - * needs to be closed. The raw results will be mapped to the given entity class and are returned as stream. The name - * of the inputCollection is derived from the inputType of the aggregation. - *
+ *

+ * Returns a {@link Stream} that wraps the Mongo DB {@link com.mongodb.client.AggregateIterable} that needs to be + * closed. The raw results will be mapped to the given entity class and are returned as stream. The name of the + * inputCollection is derived from the inputType of the aggregation. + *

* Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling * explanation mode will throw an {@link IllegalArgumentException}. * * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be * {@literal null}. * @param outputType The parametrized type of the returned list, must not be {@literal null}. - * @return The results of the aggregation operation. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). * @since 2.0 */ - CloseableIterator aggregateStream(TypedAggregation aggregation, Class outputType); + Stream aggregateStream(TypedAggregation aggregation, Class outputType); /** * Execute an aggregation operation backed by a Mongo DB {@link com.mongodb.client.AggregateIterable}. - *
- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.AggregateIterable} that - * needs to be closed. The raw results will be mapped to the given entity class. - *
+ *

+ * Returns a {@link Stream} that wraps the Mongo DB {@link com.mongodb.client.AggregateIterable} that needs to be + * closed. The raw results will be mapped to the given entity class. + *

* Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling * explanation mode will throw an {@link IllegalArgumentException}. * @@ -569,17 +572,18 @@ public interface MongoOperations extends FluentMongoOperations { * @param inputType the inputType where the aggregation operation will read from, must not be {@literal null} or * empty. * @param outputType The parametrized type of the returned list, must not be {@literal null}. - * @return The results of the aggregation operation. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). * @since 2.0 */ - CloseableIterator aggregateStream(Aggregation aggregation, Class inputType, Class outputType); + Stream aggregateStream(Aggregation aggregation, Class inputType, Class outputType); /** * Execute an aggregation operation backed by a Mongo DB {@link com.mongodb.client.AggregateIterable}. - *
- * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link com.mongodb.client.AggregateIterable} that - * needs to be closed. The raw results will be mapped to the given entity class. - *
+ *

+ * Returns a {@link Stream} that wraps the Mongo DB {@link com.mongodb.client.AggregateIterable} that needs to be + * closed. The raw results will be mapped to the given entity class. + *

* Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling * explanation mode will throw an {@link IllegalArgumentException}. * @@ -588,10 +592,11 @@ public interface MongoOperations extends FluentMongoOperations { * @param collectionName the collection where the aggregation operation will read from, must not be {@literal null} or * empty. * @param outputType The parametrized type of the returned list, must not be {@literal null}. - * @return The results of the aggregation operation. + * @return the result {@link Stream}, containing mapped objects, needing to be closed once fully processed (e.g. + * through a try-with-resources clause). * @since 2.0 */ - CloseableIterator aggregateStream(Aggregation aggregation, String collectionName, Class outputType); + Stream aggregateStream(Aggregation aggregation, String collectionName, Class outputType); /** * Execute a map-reduce operation. The map-reduce operation will be formed with an output type of INLINE diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java index bf8386a2d..8fb3310ae 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2021 the original author or authors. + * Copyright 2010-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ import java.math.RoundingMode; import java.util.*; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -375,17 +376,17 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, } @Override - public CloseableIterator stream(Query query, Class entityType) { + public Stream stream(Query query, Class entityType) { return stream(query, entityType, getCollectionName(entityType)); } @Override - public CloseableIterator stream(Query query, Class entityType, String collectionName) { + public Stream stream(Query query, Class entityType, String collectionName) { return doStream(query, entityType, collectionName, entityType); } @SuppressWarnings("ConstantConditions") - protected CloseableIterator doStream(Query query, Class entityType, String collectionName, + protected Stream doStream(Query query, Class entityType, String collectionName, Class returnType) { Assert.notNull(query, "Query must not be null!"); @@ -393,7 +394,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, Assert.hasText(collectionName, "Collection name must not be null or empty!"); Assert.notNull(returnType, "ReturnType must not be null!"); - return execute(collectionName, (CollectionCallback>) collection -> { + return execute(collectionName, (CollectionCallback>) collection -> { MongoPersistentEntity persistentEntity = mappingContext.getPersistentEntity(entityType); @@ -408,7 +409,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, col -> col.find(mappedQuery, Document.class).projection(mappedFields)); return new CloseableIterableCursorAdapter<>(cursor, exceptionTranslator, - new ProjectingReadCallback<>(mongoConverter, projection, collectionName)); + new ProjectingReadCallback<>(mongoConverter, projection, collectionName)).stream(); }); } @@ -1860,7 +1861,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, } @Override - public CloseableIterator aggregateStream(TypedAggregation aggregation, String inputCollectionName, + public Stream aggregateStream(TypedAggregation aggregation, String inputCollectionName, Class outputType) { Assert.notNull(aggregation, "Aggregation pipeline must not be null!"); @@ -1871,19 +1872,19 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, } @Override - public CloseableIterator aggregateStream(TypedAggregation aggregation, Class outputType) { + public Stream aggregateStream(TypedAggregation aggregation, Class outputType) { return aggregateStream(aggregation, getCollectionName(aggregation.getInputType()), outputType); } @Override - public CloseableIterator aggregateStream(Aggregation aggregation, Class inputType, Class outputType) { + public Stream aggregateStream(Aggregation aggregation, Class inputType, Class outputType) { return aggregateStream(aggregation, getCollectionName(inputType), outputType, new TypeBasedAggregationOperationContext(inputType, mappingContext, queryMapper)); } @Override - public CloseableIterator aggregateStream(Aggregation aggregation, String collectionName, Class outputType) { + public Stream aggregateStream(Aggregation aggregation, String collectionName, Class outputType) { return aggregateStream(aggregation, collectionName, outputType, null); } @@ -2021,7 +2022,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, } @SuppressWarnings("ConstantConditions") - protected CloseableIterator aggregateStream(Aggregation aggregation, String collectionName, + protected Stream aggregateStream(Aggregation aggregation, String collectionName, Class outputType, @Nullable AggregationOperationContext context) { Assert.hasText(collectionName, "Collection name must not be null or empty!"); @@ -2041,7 +2042,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, ReadDocumentCallback readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName); - return execute(collectionName, (CollectionCallback>) collection -> { + return execute(collectionName, (CollectionCallback>) collection -> { AggregateIterable cursor = collection.aggregate(pipeline, Document.class) // .allowDiskUse(options.isAllowDiskUse()); @@ -2061,7 +2062,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, .map(Collation::toMongoCollation) // .ifPresent(cursor::collation); - return new CloseableIterableCursorAdapter<>(cursor, exceptionTranslator, readCallback); + return new CloseableIterableCursorAdapter<>(cursor, exceptionTranslator, readCallback).stream(); }); } @@ -3203,6 +3204,23 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, } } + /** + * @deprecated since 3.1.4. Use {@link #getMongoDatabaseFactory()} instead. + * @return the {@link MongoDatabaseFactory} in use. + */ + @Deprecated + public MongoDatabaseFactory getMongoDbFactory() { + return getMongoDatabaseFactory(); + } + + /** + * @return the {@link MongoDatabaseFactory} in use. + * @since 3.1.4 + */ + public MongoDatabaseFactory getMongoDatabaseFactory() { + return mongoDbFactory; + } + /** * A {@link CloseableIterator} that is backed by a MongoDB {@link MongoCollection}. * @@ -3286,23 +3304,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, } } - /** - * @deprecated since 3.1.4. Use {@link #getMongoDatabaseFactory()} instead. - * @return the {@link MongoDatabaseFactory} in use. - */ - @Deprecated - public MongoDatabaseFactory getMongoDbFactory() { - return getMongoDatabaseFactory(); - } - - /** - * @return the {@link MongoDatabaseFactory} in use. - * @since 3.1.4 - */ - public MongoDatabaseFactory getMongoDatabaseFactory() { - return mongoDbFactory; - } - /** * {@link MongoTemplate} extension bound to a specific {@link ClientSession} that is applied when interacting with the * server through the driver API. diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedAggregation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedAggregation.java index 713ce308a..9e4f37312 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedAggregation.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedAggregation.java @@ -120,7 +120,7 @@ public class StringBasedAggregation extends AbstractMongoQuery { if (method.isStreamQuery()) { - Stream stream = mongoOperations.aggregateStream(aggregation, targetType).stream(); + Stream stream = mongoOperations.aggregateStream(aggregation, targetType); if (isSimpleReturnType) { return stream.map(it -> AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter)); diff --git a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt index e79aa5520..602558df6 100644 --- a/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt +++ b/spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/MongoOperationsExtensions.kt @@ -30,7 +30,7 @@ import org.springframework.data.mongodb.core.query.Criteria import org.springframework.data.mongodb.core.query.NearQuery import org.springframework.data.mongodb.core.query.Query import org.springframework.data.mongodb.core.query.Update -import org.springframework.data.util.CloseableIterator +import java.util.stream.Stream import kotlin.reflect.KClass /** @@ -67,8 +67,8 @@ inline fun MongoOperations.execute(action: CollectionCallback< * @author Sebastien Deleuze * @since 2.0 */ -inline fun MongoOperations.stream(query: Query): CloseableIterator = - stream(query, T::class.java) +inline fun MongoOperations.stream(query: Query): Stream = + stream(query, T::class.java) /** * Extension for [MongoOperations.stream] leveraging reified type parameters. @@ -76,9 +76,12 @@ inline fun MongoOperations.stream(query: Query): CloseableIter * @author Sebastien Deleuze * @since 2.0 */ -inline fun MongoOperations.stream(query: Query, collectionName: String? = null): CloseableIterator = - if (collectionName != null) stream(query, T::class.java, collectionName) - else stream(query, T::class.java) +inline fun MongoOperations.stream( + query: Query, + collectionName: String? = null +): Stream = + if (collectionName != null) stream(query, T::class.java, collectionName) + else stream(query, T::class.java) /** * Extension for [MongoOperations.createCollection] providing a [KClass] based variant. @@ -264,7 +267,7 @@ inline fun MongoOperations.aggregate( inline fun MongoOperations.aggregateStream( aggregation: Aggregation, inputType: KClass<*> -): CloseableIterator = +): Stream = aggregateStream(aggregation, inputType.java, O::class.java) /** @@ -273,7 +276,7 @@ inline fun MongoOperations.aggregateStream( * @author Mark Paluch * @since 3.2 */ -inline fun MongoOperations.aggregateStream(aggregation: Aggregation): CloseableIterator = +inline fun MongoOperations.aggregateStream(aggregation: Aggregation): Stream = aggregateStream(aggregation, I::class.java, O::class.java) /** @@ -285,7 +288,7 @@ inline fun MongoOperations.aggregateStream(ag inline fun MongoOperations.aggregateStream( aggregation: Aggregation, collectionName: String -): CloseableIterator = +): Stream = aggregateStream(aggregation, collectionName, O::class.java) /** diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java index a50c6e625..5a2e7e72b 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java @@ -40,6 +40,7 @@ import java.util.*; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.IntStream; +import java.util.stream.Stream; import org.bson.types.ObjectId; import org.junit.jupiter.api.AfterEach; @@ -85,7 +86,6 @@ import org.springframework.data.mongodb.test.util.Client; import org.springframework.data.mongodb.test.util.MongoClientExtension; import org.springframework.data.mongodb.test.util.MongoTestTemplate; import org.springframework.data.mongodb.test.util.MongoVersion; -import org.springframework.data.util.CloseableIterator; import org.springframework.test.annotation.DirtiesContext; import org.springframework.util.ClassUtils; import org.springframework.util.ObjectUtils; @@ -3044,10 +3044,9 @@ public class MongoTemplateTests { Query q = new Query(); q.with(Sort.by(Direction.ASC, "age")); - CloseableIterator stream = template.stream(q, Person.class); + List streamResults = template.stream(q, Person.class).map(Person::getAge).toList(); - assertThat(stream.next().getAge()).isEqualTo(youngestPerson.getAge()); - assertThat(stream.next().getAge()).isEqualTo(oldestPerson.getAge()); + assertThat(streamResults).containsExactly(youngestPerson.getAge(), oldestPerson.getAge()); } @Test // DATAMONGO-1208 @@ -3060,7 +3059,7 @@ public class MongoTemplateTests { Query q = new Query(); q.with(PageRequest.of(0, 1, Sort.by(Direction.ASC, "age"))); - CloseableIterator stream = template.stream(q, Person.class); + Iterator stream = template.stream(q, Person.class).iterator(); assertThat(stream.next().getAge()).isEqualTo(youngestPerson.getAge()); assertThat(stream.hasNext()).isFalse(); @@ -3332,11 +3331,11 @@ public class MongoTemplateTests { template.insert(document, "some_special_collection"); - CloseableIterator stream = template.stream(new Query(), Document.class); - assertThat(stream.hasNext()).isFalse(); + Stream stream = template.stream(new Query(), Document.class); + assertThat(stream).isEmpty(); - CloseableIterator stream2 = template.stream(new Query(where("_id").is(document.id)), - org.bson.Document.class, "some_special_collection"); + Iterator stream2 = template + .stream(new Query(where("_id").is(document.id)), org.bson.Document.class, "some_special_collection").iterator(); assertThat(stream2.hasNext()).isTrue(); assertThat(stream2.next().get("_id")).isEqualTo(new ObjectId(document.id)); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java index 6c65f4eef..0a729f673 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java @@ -803,7 +803,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests { @Test // DATAMONGO-1518 void streamQueryShouldUseCollationWhenPresent() { - template.stream(new BasicQuery("{}").collation(Collation.of("fr")), AutogenerateableId.class).next(); + template.stream(new BasicQuery("{}").collation(Collation.of("fr")), AutogenerateableId.class); verify(findIterable).collation(eq(com.mongodb.client.model.Collation.builder().locale("fr").build())); } @@ -1221,7 +1221,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests { @Test // DATAMONGO-1854 void streamQueryShouldUseDefaultCollationWhenPresent() { - template.stream(new BasicQuery("{}"), Sith.class).next(); + template.stream(new BasicQuery("{}"), Sith.class); verify(findIterable).collation(eq(com.mongodb.client.model.Collation.builder().locale("de_AT").build())); } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java index d65cdf63f..b7663ba3a 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java @@ -40,6 +40,7 @@ import java.util.Collections; import java.util.Date; import java.util.List; import java.util.Scanner; +import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -49,6 +50,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; + import org.springframework.core.io.ClassPathResource; import org.springframework.dao.DataAccessException; import org.springframework.data.annotation.Id; @@ -63,7 +65,6 @@ import org.springframework.data.mongodb.core.TestEntities; import org.springframework.data.mongodb.core.Venue; import org.springframework.data.mongodb.core.aggregation.AggregationTests.CarDescriptor.Entry; import org.springframework.data.mongodb.core.aggregation.BucketAutoOperation.Granularities; -import org.springframework.data.mongodb.core.aggregation.DateOperators.TemporalUnits; import org.springframework.data.mongodb.core.aggregation.VariableOperators.Let.ExpressionVariable; import org.springframework.data.mongodb.core.geo.GeoJsonPoint; import org.springframework.data.mongodb.core.index.GeoSpatialIndexType; @@ -76,7 +77,6 @@ import org.springframework.data.mongodb.test.util.MongoTemplateExtension; import org.springframework.data.mongodb.test.util.MongoTestTemplate; import org.springframework.data.mongodb.test.util.MongoVersion; import org.springframework.data.mongodb.test.util.Template; -import org.springframework.data.util.CloseableIterator; import com.mongodb.MongoException; import com.mongodb.client.MongoCollection; @@ -234,18 +234,17 @@ public class AggregationTests { sort(DESC, "n") // ).withOptions(new AggregationOptions(true, false, 1)); - CloseableIterator iterator = mongoTemplate.aggregateStream(agg, INPUT_COLLECTION, TagCount.class); + try (Stream stream = mongoTemplate.aggregateStream(agg, INPUT_COLLECTION, TagCount.class)) { - assertThat(iterator).isNotNull(); - List tagCount = toList(iterator); - iterator.close(); + List tagCount = stream.toList(); - assertThat(tagCount).isNotNull(); - assertThat(tagCount.size()).isEqualTo(3); + assertThat(tagCount).isNotNull(); + assertThat(tagCount.size()).isEqualTo(3); - assertTagCount("spring", 3, tagCount.get(0)); - assertTagCount("mongodb", 2, tagCount.get(1)); - assertTagCount("nosql", 1, tagCount.get(2)); + assertTagCount("spring", 3, tagCount.get(0)); + assertTagCount("mongodb", 2, tagCount.get(1)); + assertTagCount("nosql", 1, tagCount.get(2)); + } } @Test // DATAMONGO-586 @@ -284,14 +283,12 @@ public class AggregationTests { sort(DESC, "n") // ); - CloseableIterator results = mongoTemplate.aggregateStream(aggregation, INPUT_COLLECTION, TagCount.class); + try (Stream stream = mongoTemplate.aggregateStream(aggregation, INPUT_COLLECTION, TagCount.class)) { - assertThat(results).isNotNull(); + List tagCount = stream.toList(); - List tagCount = toList(results); - results.close(); - - assertThat(tagCount.size()).isEqualTo(0); + assertThat(tagCount.size()).isEqualTo(0); + } } @Test // DATAMONGO-1391 @@ -384,16 +381,14 @@ public class AggregationTests { limit(2) // ); - CloseableIterator results = mongoTemplate.aggregateStream(aggregation, INPUT_COLLECTION, TagCount.class); + try (Stream stream = mongoTemplate.aggregateStream(aggregation, INPUT_COLLECTION, TagCount.class)) { - assertThat(results).isNotNull(); + List tagCount = stream.toList(); - List tagCount = toList(results); - results.close(); - - assertThat(tagCount.size()).isEqualTo(2); - assertTagCount(null, 0, tagCount.get(0)); - assertTagCount(null, 0, tagCount.get(1)); + assertThat(tagCount.size()).isEqualTo(2); + assertTagCount(null, 0, tagCount.get(0)); + assertTagCount(null, 0, tagCount.get(1)); + } } @Test // DATAMONGO-586 @@ -1280,19 +1275,18 @@ public class AggregationTests { assertThat(agg).isNotNull(); assertThat(agg.toString()).isNotNull(); - CloseableIterator iterator = mongoTemplate.aggregateStream(agg, LikeStats.class); - List result = toList(iterator); - iterator.close(); + try (Stream stream = mongoTemplate.aggregateStream(agg, LikeStats.class)) { - assertThat(result).isNotNull(); - assertThat(result).isNotNull(); - assertThat(result.size()).isEqualTo(5); + List result = stream.toList(); - assertLikeStats(result.get(0), "a", 4); - assertLikeStats(result.get(1), "b", 2); - assertLikeStats(result.get(2), "c", 4); - assertLikeStats(result.get(3), "d", 2); - assertLikeStats(result.get(4), "e", 3); + assertThat(result.size()).isEqualTo(5); + + assertLikeStats(result.get(0), "a", 4); + assertLikeStats(result.get(1), "b", 2); + assertLikeStats(result.get(2), "c", 4); + assertLikeStats(result.get(3), "d", 2); + assertLikeStats(result.get(4), "e", 3); + } } @Test // DATAMONGO-960 @@ -1606,13 +1600,14 @@ public class AggregationTests { sort(DESC, "count"), // out(tempOutCollection)); - CloseableIterator iterator = mongoTemplate.aggregateStream(agg, Document.class); + try (Stream stream = mongoTemplate.aggregateStream(agg, Document.class)) { - List result = toList(iterator); + List result = stream.toList(); - assertThat(result).hasSize(2); - assertThat(result.get(0)).containsEntry("_id", "MALE").containsEntry("count", 3); - assertThat(result.get(1)).containsEntry("_id", "FEMALE").containsEntry("count", 2); + assertThat(result).hasSize(2); + assertThat(result.get(0)).containsEntry("_id", "MALE").containsEntry("count", 3); + assertThat(result.get(1)).containsEntry("_id", "FEMALE").containsEntry("count", 2); + } mongoTemplate.dropCollection(tempOutCollection); } @@ -2013,16 +2008,6 @@ public class AggregationTests { assertThat(tagCount.getN()).isEqualTo(n); } - private static List toList(CloseableIterator results) { - - List result = new ArrayList(); - while (results.hasNext()) { - result.add(results.next()); - } - - return result; - } - static class DATAMONGO753 { PD[] pd; diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedAggregationUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedAggregationUnitTests.java index 995442f0f..4918f73ed 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedAggregationUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedAggregationUnitTests.java @@ -42,7 +42,6 @@ import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Slice; -import org.springframework.data.domain.SliceImpl; import org.springframework.data.domain.Sort; import org.springframework.data.domain.Sort.Direction; import org.springframework.data.mongodb.InvalidMongoDbApiUsageException; @@ -66,7 +65,6 @@ import org.springframework.data.projection.SpelAwareProxyProjectionFactory; import org.springframework.data.repository.Repository; import org.springframework.data.repository.core.support.DefaultRepositoryMetadata; import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider; -import org.springframework.data.util.CloseableIterator; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.lang.Nullable; import org.springframework.util.ClassUtils; @@ -238,22 +236,7 @@ public class StringBasedAggregationUnitTests { @Test // GH-3543 void aggregationWithStreamReturnType() { - when(operations.aggregateStream(any(TypedAggregation.class), any())).thenReturn(new CloseableIterator() { - @Override - public void close() { - - } - - @Override - public boolean hasNext() { - return false; - } - - @Override - public Object next() { - return null; - } - }); + when(operations.aggregateStream(any(TypedAggregation.class), any())).thenReturn(Stream.empty()); StringBasedAggregation sba = createAggregationForMethod("aggregationWithStreamReturnType", Pageable.class);