DATAMONGO-1165 - Polishing.

Renamed MongoOperations executeAsStream(…) to stream(…). Make use of Spring Data Commons StreamUtils in AbstractMongoQuery's StreamExecution. Moved test case from PersonRepositoryIntegrationTests to AbstractPersonRepositoryIntegrationTests to make sure they're executed for all sub-types.

Original pull request: #274.

Author: odrotbohm

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

@@ -179,7 +179,7 @@ public interface MongoOperations {
 	 * @return
 	 * @since 1.7
 	 */
-	<T> CloseableIterator<T> executeAsStream(Query query, Class<T> entityType);
+	<T> CloseableIterator<T> stream(Query query, Class<T> entityType);
 
 	/**
 	 * Create an uncapped collection with a name based on the provided entity class.

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

@@ -95,9 +95,8 @@
 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 org.springframework.data.mongodb.util.CloseableIterator;
 import org.springframework.data.mongodb.util.MongoClientVersion;
+import org.springframework.data.util.CloseableIterator;
 import org.springframework.jca.cci.core.ConnectionCallback;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -318,11 +317,12 @@ public MongoConverter getConverter() {
 		return this.mongoConverter;
 	}
 
-	/* (non-Javadoc)
+	/* 
+	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.MongoOperations#executeAsStream(org.springframework.data.mongodb.core.query.Query, java.lang.Class)
 	 */
 	@Override
-	public <T> CloseableIterator<T> executeAsStream(final Query query, final Class<T> entityType) {
+	public <T> CloseableIterator<T> stream(final Query query, final Class<T> entityType) {
 
 		return execute(entityType, new CollectionCallback<CloseableIterator<T>>() {
 
@@ -340,7 +340,6 @@ public CloseableIterator<T> doInCollection(DBCollection collection) throws Mongo
 				return new CloseableIterableCusorAdapter<T>(cursor, exceptionTranslator, readCallback);
 			}
 		});
-
 	}
 
 	public String getCollectionName(Class<?> entityClass) {

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

@@ -17,9 +17,6 @@
 
 import java.util.Collections;
 import java.util.List;
-import java.util.Spliterator;
-import java.util.Spliterators;
-import java.util.stream.StreamSupport;
 
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
@@ -36,7 +33,7 @@
 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.StreamUtils;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
 
@@ -116,6 +113,7 @@ private Query applyQueryMetaAttributesWhenPresent(Query query) {
 		if (method.hasQueryMetaAttributes()) {
 			query.setMeta(method.getQueryMetaAttributes());
 		}
+
 		return query;
 	}
 
@@ -425,19 +423,17 @@ private Object deleteAndConvertResult(Query query, MongoEntityMetadata<?> metada
 	 */
 	final class StreamExecution extends Execution {
 
-		/* (non-Javadoc)
+		/* 
+		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery.Execution#execute(org.springframework.data.mongodb.core.query.Query)
 		 */
 		@Override
+		@SuppressWarnings("unchecked")
 		Object execute(Query query) {
 
 			Class<?> entityType = getQueryMethod().getEntityInformation().getJavaType();
 
-			@SuppressWarnings("unchecked")
-			CloseableIterator<Object> result = (CloseableIterator<Object>) operations.executeAsStream(query, entityType);
-			Spliterator<Object> spliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.NONNULL);
-
-			return StreamSupport.stream(spliterator, false).onClose(new CloseableIteratorDisposingRunnable(result));
+			return StreamUtils.createStreamFromIterator((CloseableIterator<Object>) operations.stream(query, entityType));
 		}
 	}
 }

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

@@ -23,6 +23,8 @@
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.hamcrest.Matchers;
 import org.junit.Before;
@@ -65,7 +67,7 @@ public abstract class AbstractPersonRepositoryIntegrationTests {
 
 	@Autowired MongoOperations operations;
 
-	protected Person dave, oliver, carter, boyd, stefan, leroi, alicia;
+	Person dave, oliver, carter, boyd, stefan, leroi, alicia;
 	QPerson person;
 
 	List<Person> all;
@@ -1148,4 +1150,19 @@ public void shouldSupportSortingWithQSort() throws Exception {
 		assertThat(result, is(Matchers.<Person> iterableWithSize(persons.size())));
 		assertThat(result.iterator().next().getFirstname(), is("Siggi 2"));
 	}
+
+	/**
+	 * @see DATAMONGO-1165
+	 */
+	@Test
+	public void shouldAllowReturningJava8StreamInCustomQuery() throws Exception {
+
+		Stream<Person> result = repository.findByCustomQueryWithStreamingCursorByFirstnames(Arrays.asList("Dave"));
+
+		try {
+			assertThat(result.collect(Collectors.<Person> toList()), hasItems(dave));
+		} finally {
+			result.close();
+		}
+	}
 }

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

@@ -325,6 +325,6 @@ public interface PersonRepository extends MongoRepository<Person, String>, Query
 	/**
 	 * @see DATAMONGO-1165
 	 */
-	@Query("{firstname:{$in:?0}}")
+	@Query("{ firstname : { $in : ?0 }}")
 	Stream<Person> findByCustomQueryWithStreamingCursorByFirstnames(List<String> firstnames);
 }

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

@@ -15,15 +15,6 @@
  */
 package org.springframework.data.mongodb.repository;
 
-import static org.hamcrest.Matchers.*;
-import static org.junit.Assert.*;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import org.junit.Test;
 import org.springframework.test.context.ContextConfiguration;
 
 /**
@@ -33,21 +24,4 @@
  * @author Thomas Darimont
  */
 @ContextConfiguration
-public class PersonRepositoryIntegrationTests extends AbstractPersonRepositoryIntegrationTests {
-
-	/**
-	 * @see DATAMONGO-1165
-	 */
-	@Test
-	public void shouldAllowReturningJava8StreamInCustomQuery() throws Exception {
-
-		Stream<Person> result = repository.findByCustomQueryWithStreamingCursorByFirstnames(Arrays.asList("Dave"));
-
-		try {
-			List<Person> readPersons = result.collect(Collectors.<Person> toList());
-			assertThat(readPersons, hasItems(dave));
-		} finally {
-			result.close();
-		}
-	}
-}
+public class PersonRepositoryIntegrationTests extends AbstractPersonRepositoryIntegrationTests {}