DATAMONGO-1208 - Use QueryCursorPreparer for streaming in MongoTemplate.

We now use the QueryCursorPreparer honor skip, limit, sort, etc. for streaming.

Original pull request: #297.
Polishing pull request: #298.

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

@@ -136,6 +136,7 @@ import com.mongodb.util.JSONParseException;
  * @author Thomas Darimont
  * @author Chuong Ngo
  * @author Christoph Strobl
+ * @author Doménique Tilleuil
  */
 @SuppressWarnings("deprecation")
 public class MongoTemplate implements MongoOperations, ApplicationContextAware {
@@ -335,9 +336,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware {
 				DBObject mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), persistentEntity);
 
 				DBCursor cursor = collection.find(mappedQuery, mappedFields);
+				QueryCursorPreparer cursorPreparer = new QueryCursorPreparer(query, entityType);
+
 				ReadDbObjectCallback<T> readCallback = new ReadDbObjectCallback<T>(mongoConverter, entityType);
 
-				return new CloseableIterableCusorAdapter<T>(cursor, exceptionTranslator, readCallback);
+				return new CloseableIterableCusorAdapter<T>(cursorPreparer.prepare(cursor), exceptionTranslator, readCallback);
 			}
 		});
 	}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java

@@ -73,6 +73,7 @@ import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Criteria;
 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.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ObjectUtils;
@@ -2742,6 +2743,44 @@ public class MongoTemplateTests {
 		assertThat(result, hasItems(newYork, washington));
 	}
 
+	/**
+	 * @see DATAMONGO-1208
+	 */
+	@Test
+	public void takesSortIntoAccountWhenStreaming() {
+
+		Person youngestPerson = new Person("John", 20);
+		Person oldestPerson = new Person("Jane", 42);
+
+		template.insertAll(Arrays.asList(oldestPerson, youngestPerson));
+
+		Query q = new Query();
+		q.with(new Sort(Direction.ASC, "age"));
+		CloseableIterator<Person> stream = template.stream(q, Person.class);
+
+		assertThat(stream.next().getAge(), is(youngestPerson.getAge()));
+		assertThat(stream.next().getAge(), is(oldestPerson.getAge()));
+	}
+
+	/**
+	 * @see DATAMONGO-1208
+	 */
+	@Test
+	public void takesLimitIntoAccountWhenStreaming() {
+		
+		Person youngestPerson = new Person("John", 20);
+		Person oldestPerson = new Person("Jane", 42);
+
+		template.insertAll(Arrays.asList(oldestPerson, youngestPerson));
+
+		Query q = new Query();
+		q.with(new PageRequest(0, 1, new Sort(Direction.ASC, "age")));
+		CloseableIterator<Person> stream = template.stream(q, Person.class);
+
+		assertThat(stream.next().getAge(), is(youngestPerson.getAge()));
+		assertThat(stream.hasNext(), is(false));
+	}
+
 	static class DoucmentWithNamedIdField {
 
 		@Id String someIdKey;