DATAMONGO-2080 - Use fluent API for reactive tailable query methods.

Using the fluent API allows using DTO projections with properties that are unknown to the actual domain object. Previously, DTO projections attempted to read the domain type and during property access, missing properties were reported with an IllegalArgumentException. Unknown properties remain now unset.

Original Pull Request: #608

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

@@ -32,7 +32,6 @@ import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecu
 import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecution.GeoNearExecution;
 import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecution.ResultProcessingConverter;
 import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecution.ResultProcessingExecution;
-import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryExecution.TailExecution;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
@@ -146,7 +145,7 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 		} else if (method.isGeoNearQuery()) {
 			return new GeoNearExecution(operations, accessor, method.getReturnType());
 		} else if (isTailable(method)) {
-			return new TailExecution(operations, accessor.getPageable());
+			return (q, t, c) -> operation.matching(q.with(accessor.getPageable())).tail();
 		} else if (method.isCollectionQuery()) {
 			return (q, t, c) -> operation.matching(q.with(accessor.getPageable())).all();
 		} else if (isCountQuery()) {

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

@@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.offset;
 import static org.springframework.data.domain.Sort.Direction.*;
 import static org.springframework.data.mongodb.test.util.Assertions.assertThat;
 
+import lombok.Data;
 import lombok.NoArgsConstructor;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
@@ -227,6 +228,19 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF
 		disposable.dispose();
 	}
 
+	@Test // DATAMONGO-2080
+	public void shouldUseTailableCursorWithDtoProjection() {
+
+		StepVerifier.create(template.dropCollection(Capped.class) //
+				.then(template.createCollection(Capped.class, //
+						CollectionOptions.empty().size(1000).maxDocuments(100).capped()))) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		StepVerifier.create(template.insert(new Capped("value", Math.random()))).expectNextCount(1).verifyComplete();
+		StepVerifier.create(cappedRepository.findDtoProjectionByKey("value")).expectNextCount(1).thenCancel().verify();
+	}
+
 	@Test // DATAMONGO-1444
 	public void findsPeopleByLocationWithinCircle() {
 
@@ -337,6 +351,8 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF
 
 		Mono<Person> findOneByLastname(String lastname);
 
+		Mono<DtoProjection> findOneProjectedByLastname(String lastname);
+
 		Mono<Person> findByLastname(Publisher<String> lastname);
 
 		Flux<Person> findByLastnameIn(Publisher<String> lastname);
@@ -376,6 +392,9 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF
 
 		@Tailable
 		Flux<CappedProjection> findProjectionByKey(String key);
+
+		@Tailable
+		Flux<DtoProjection> findDtoProjectionByKey(String key);
 	}
 
 	@Document
@@ -395,4 +414,10 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF
 	interface CappedProjection {
 		double getRandom();
 	}
+
+	@Data
+	static class DtoProjection {
+		String id;
+		double unknown;
+	}
 }