DATAMONGO-1165 - Add support for Streaming large result lists.

Initial support for CloteableIterable and Java 8 Stream as return type of custom Repository finder methods.

Author: Thomas Darimont

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java

@@ -20,11 +20,14 @@
 import java.util.Set;
 
 import org.springframework.data.geo.GeoResults;
+import org.springframework.data.mapping.context.MappingContext;
 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.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 import org.springframework.data.mongodb.core.mapreduce.GroupBy;
 import org.springframework.data.mongodb.core.mapreduce.GroupByResults;
 import org.springframework.data.mongodb.core.mapreduce.MapReduceOptions;
@@ -967,4 +970,12 @@ <T> T findAndModify(Query query, Update update, FindAndModifyOptions options, Cl
 	 * @return
 	 */
 	QueryMapper getQueryMapper();
+
+	/**
+	 * Returns the underlying {@link MappingContext}.
+	 * 
+	 * @return
+	 * @since 1.7
+	 */
+	MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> getMappingContext();
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -320,6 +320,14 @@ public QueryMapper getQueryMapper() {
 		return queryMapper;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.MongoOperations#getMappingContext()
+	 */
+	@Override
+	public MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> getMappingContext() {
+		return mappingContext;
+	}
+
 	public String getCollectionName(Class<?> entityClass) {
 		return this.determineCollectionName(entityClass);
 	}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/QueryMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2015 the original author or authors.
+ * Copyright 2011-2014 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.
@@ -81,23 +81,6 @@ public QueryMapper(MongoConverter converter) {
 		this.mappingContext = converter.getMappingContext();
 	}
 
-	/**
-	 * Replaces the property keys used in the given {@link DBObject} with the appropriate keys by using the
-	 * {@link PersistentEntity} metadata linked with the given {@code entityClass}.
-	 * 
-	 * @see #getMappedObject(DBObject, MongoPersistentEntity)
-	 * @param query must not be {@literal null}.
-	 * @param entityClass must not be {@literal null}.
-	 * @return
-	 * @since 1.7
-	 */
-	public DBObject getMappedObject(DBObject query, Class<?> entityClass) {
-
-		Assert.notNull(entityClass, "Entity class must not be null!");
-
-		return getMappedObject(query, mappingContext.getPersistentEntity(entityClass));
-	}
-
 	/**
 	 * Replaces the property keys used in the given {@link DBObject} with the appropriate keys by using the
 	 * {@link PersistentEntity} metadata.
@@ -161,26 +144,9 @@ public DBObject getMappedSort(DBObject sortObject, MongoPersistentEntity<?> enti
 		return mappedSort;
 	}
 
-	/**
-	 * Maps fields to retrieve to the {@link MongoPersistentEntity}s properties linked to the given {@code entityClass}.
-	 * <p>
-	 * 
-	 * @see QueryMapper#getMappedFields(DBObject, MongoPersistentEntity)
-	 * @param fieldsObject
-	 * @param entityClass must not be {@literal null}
-	 * @return
-	 * @since 1.7
-	 */
-	public DBObject getMappedFields(DBObject fieldsObject, Class<?> entityClass) {
-
-		Assert.notNull(entityClass, "Entity class must not be null!");
-
-		return getMappedFields(fieldsObject, mappingContext.getPersistentEntity(entityClass));
-	}
-
 	/**
 	 * Maps fields to retrieve to the {@link MongoPersistentEntity}s properties. <br />
-	 * Also converts and potentially adds missing property {@code $meta} representation.
+	 * Also onverts and potentially adds missing property {@code $meta} representation.
 	 * 
 	 * @param fieldsObject
 	 * @param entity
@@ -317,7 +283,7 @@ protected Object getMappedValue(Field documentField, Object value) {
 				} else if (valueDbo.containsField("$ne")) {
 					resultDbo.put("$ne", convertId(valueDbo.get("$ne")));
 				} else {
-					return getMappedObject(resultDbo, (MongoPersistentEntity<?>) null);
+					return getMappedObject(resultDbo, null);
 				}
 
 				return resultDbo;

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsTemplate.java

@@ -26,7 +26,6 @@
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
-import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -258,7 +257,7 @@ private DBObject getMappedQuery(Query query) {
 	}
 
 	private DBObject getMappedQuery(DBObject query) {
-		return query == null ? null : queryMapper.getMappedObject(query, (MongoPersistentEntity<?>) null);
+		return query == null ? null : queryMapper.getMappedObject(query, null);
 	}
 
 	private GridFS getGridFs() {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java

@@ -17,6 +17,9 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.StreamSupport;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.domain.PageImpl;
@@ -31,16 +34,19 @@
 import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.util.CloseableIterableCusorAdapter;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.util.CloseableIterator;
+import org.springframework.data.util.CloseableIteratorDisposingRunnable;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
 
 import com.mongodb.DBCollection;
+import com.mongodb.DBCursor;
 import com.mongodb.DBObject;
 import com.mongodb.MongoException;
 import com.mongodb.WriteResult;
@@ -91,11 +97,11 @@ public Object execute(Object[] parameters) {
 
 		applyQueryMetaAttributesWhenPresent(query);
 
-		if (method.isStreamQuery()) {
+		if (method.isCloseableIteratorQuery()) {
+			return new CursorBackedExecution().execute(query);
+		} else if (method.isStreamQuery()) {
 			return new StreamExecution().execute(query);
-		}
-
-		if (isDeleteQuery()) {
+		} else if (isDeleteQuery()) {
 			return new DeleteExecution().execute(query);
 		} else if (method.isGeoNearQuery() && method.isPageQuery()) {
 
@@ -424,30 +430,54 @@ private Object deleteAndConvertResult(Query query, MongoEntityMetadata<?> metada
 		}
 	}
 
-	final class StreamExecution extends Execution {
+	/**
+	 * @author Thomas Darimont
+	 */
+	private class CursorBackedExecution extends Execution {
 
 		@Override
 		Object execute(final Query query) {
 
-			final Class<?> entityType = getQueryMethod().getEntityInformation().getJavaType();
-			final QueryMapper queryMapper = operations.getQueryMapper();
+			Class<?> javaType = getQueryMethod().getEntityInformation().getJavaType();
+			QueryMapper queryMapper = operations.getQueryMapper();
+
+			return createCursorAndWrapInCloseableIterator(query, javaType, queryMapper);
+		}
+
+		@SuppressWarnings("unchecked")
+		private CloseableIterator<Object> createCursorAndWrapInCloseableIterator(final Query query,
+				final Class<?> entityType, final QueryMapper queryMapper) {
+
+			final MongoPersistentEntity<?> persistentEntity = operations.getMappingContext().getPersistentEntity(entityType);
+
+			return (CloseableIterator<Object>) operations.execute(entityType, new CollectionCallback<Object>() {
 
-			@SuppressWarnings("unchecked")
-			CloseableIterator<Object> result = (CloseableIterator<Object>) operations.execute(entityType,
-					new CollectionCallback<Object>() {
+				@Override
+				public Object doInCollection(DBCollection collection) throws MongoException, DataAccessException {
 
-						@Override
-						public Object doInCollection(DBCollection collection) throws MongoException, DataAccessException {
+					DBObject mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), persistentEntity);
+					DBObject mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), persistentEntity);
 
-							DBObject mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), entityType);
-							DBObject mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entityType);
+					DBCursor cursor = collection.find(mappedQuery, mappedFields);
+
+					return new CloseableIterableCusorAdapter<Object>(cursor, operations.getConverter(), entityType);
+				}
+			});
+		}
+	}
+
+	/**
+	 * @author Thomas Darimont
+	 */
+	final class StreamExecution extends CursorBackedExecution {
+
+		@Override
+		Object execute(Query query) {
 
-							return new CloseableIterableCusorAdapter<Object>(collection.find(mappedQuery, mappedFields), operations
-									.getConverter(), entityType);
-						}
-					});
+			CloseableIterator<Object> result = (CloseableIterator<Object>) super.execute(query);
+			Spliterator<Object> spliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.NONNULL);
 
-			return result;
+			return StreamSupport.stream(spliterator, false).onClose(new CloseableIteratorDisposingRunnable(result));
 		}
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/MongoQueryMethod.java

@@ -32,7 +32,6 @@
 import org.springframework.data.repository.core.RepositoryMetadata;
 import org.springframework.data.repository.query.QueryMethod;
 import org.springframework.data.util.ClassTypeInformation;
-import org.springframework.data.util.CloseableIterator;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -234,10 +233,4 @@ public org.springframework.data.mongodb.core.query.Meta getQueryMetaAttributes()
 
 		return metaAttributes;
 	}
-
-	public boolean isStreamQuery() {
-
-		Class<?> returnType = method.getReturnType();
-		return org.springframework.util.ClassUtils.isAssignable(CloseableIterator.class, returnType);
-	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SpringDataMongodbSerializer.java

@@ -100,7 +100,7 @@ protected String getKeyForPath(Path<?> expr, PathMetadata<?> metadata) {
 	protected DBObject asDBObject(String key, Object value) {
 
 		if (ID_KEY.equals(key)) {
-			return mapper.getMappedObject(super.asDBObject(key, value), (MongoPersistentEntity<?>) null);
+			return mapper.getMappedObject(super.asDBObject(key, value), null);
 		}
 
 		return super.asDBObject(key, value instanceof Pattern ? value : converter.convertToMongoType(value));

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/CloseableIterableCusorAdapter.java

@@ -15,10 +15,10 @@
  */
 package org.springframework.data.mongodb.util;
 
-import java.io.IOException;
 import java.util.Iterator;
 
 import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.util.CloseableIterator;
 
 import com.mongodb.Cursor;
 import com.mongodb.DBObject;
@@ -72,7 +72,7 @@ public T next() {
 	}
 
 	@Override
-	public void close() throws IOException {
+	public void close() {
 
 		Cursor c = cursor;
 		try {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/AbstractPersonRepositoryIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -65,7 +65,7 @@ public abstract class AbstractPersonRepositoryIntegrationTests {
 
 	@Autowired MongoOperations operations;
 
-	Person dave, oliver, carter, boyd, stefan, leroi, alicia;
+	protected Person dave, oliver, carter, boyd, stefan, leroi, alicia;
 	QPerson person;
 
 	List<Person> all;

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepository.java

@@ -18,6 +18,7 @@
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
@@ -31,8 +32,8 @@
 import org.springframework.data.geo.Point;
 import org.springframework.data.geo.Polygon;
 import org.springframework.data.mongodb.repository.Person.Sex;
-import org.springframework.data.mongodb.util.AutoCloseableIterator;
 import org.springframework.data.querydsl.QueryDslPredicateExecutor;
+import org.springframework.data.util.CloseableIterator;
 
 /**
  * Sample repository managing {@link Person} entities.
@@ -323,5 +324,8 @@ public interface PersonRepository extends MongoRepository<Person, String>, Query
 	List<Person> findByKeyValue(String key, String value);
 
 	@Query("{firstname:{$in:?0}}")
-	AutoCloseableIterator<Person> findByCustomQueryWithCursorByFirstnames(List<String> firstnames);
+	CloseableIterator<Person> findByCustomQueryWithCursorByFirstnames(List<String> firstnames);
+
+	@Query("{firstname:{$in:?0}}")
+	Stream<Person> findByCustomQueryWithStreamingCursorByFirstnames(List<String> firstnames);
 }