DATACASS-359 - Support DTO projections on reactive repository query methods.

Original pull request: #91.

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import org.reactivestreams.Publisher;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.cassandra.core.CassandraOperations;
@@ -23,14 +26,12 @@ import org.springframework.data.cassandra.repository.query.ReactiveCassandraQuer
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.ResultProcessingConverter;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.ResultProcessingExecution;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.SingleEntityExecution;
+import org.springframework.data.convert.EntityInstantiators;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
  * Base class for reactive {@link RepositoryQuery} implementations for Cassandra.
  *
@@ -41,6 +42,7 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 
 	private final ReactiveCassandraQueryMethod method;
 	private final ReactiveCassandraOperations operations;
+	private final EntityInstantiators instantiators;
 
 	/**
 	 * Creates a new {@link AbstractReactiveCassandraQuery} from the given {@link CassandraQueryMethod} and
@@ -56,6 +58,7 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 
 		this.method = method;
 		this.operations = operations;
+		this.instantiators = new EntityInstantiators();
 	}
 
 	/* (non-Javadoc)
@@ -97,8 +100,8 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 
 		ResultProcessor resultProcessor = method.getResultProcessor().withDynamicProjection(convertingParameterAccessor);
 
-		ReactiveCassandraQueryExecution queryExecution = getExecution(query, convertingParameterAccessor,
-				new ResultProcessingConverter(resultProcessor));
+		ReactiveCassandraQueryExecution queryExecution = getExecution(convertingParameterAccessor,
+				new ResultProcessingConverter(resultProcessor, operations.getConverter().getMappingContext(), instantiators));
 
 		CassandraReturnedType returnedType = new CassandraReturnedType(resultProcessor.getReturnedType(),
 				operations.getConverter().getCustomConversions());
@@ -111,21 +114,18 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 	/**
 	 * Returns the execution instance to use.
 	 *
-	 * @param query must not be {@literal null}.
 	 * @param accessor must not be {@literal null}.
 	 * @param resultProcessing must not be {@literal null}. @return
 	 */
-	private ReactiveCassandraQueryExecution getExecution(String query, CassandraParameterAccessor accessor,
+	private ReactiveCassandraQueryExecution getExecution(CassandraParameterAccessor accessor,
 			Converter<Object, Object> resultProcessing) {
 
-		return new ResultProcessingExecution(getExecutionToWrap(accessor, resultProcessing), resultProcessing);
+		return new ResultProcessingExecution(getExecutionToWrap(), resultProcessing);
 	}
 
-	private ReactiveCassandraQueryExecution getExecutionToWrap(CassandraParameterAccessor accessor,
-			Converter<Object, Object> resultProcessing) {
+	private ReactiveCassandraQueryExecution getExecutionToWrap() {
 
-		return (method.isCollectionQuery() ? new CollectionExecution(operations)
-				: new SingleEntityExecution(operations));
+		return (method.isCollectionQuery() ? new CollectionExecution(operations) : new SingleEntityExecution(operations));
 	}
 
 	/**

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java

@@ -16,19 +16,17 @@
 
 package org.springframework.data.cassandra.repository.query;
 
-import java.util.function.Function;
-
-import org.springframework.core.convert.converter.Converter;
-import org.springframework.data.cassandra.core.CassandraOperations;
-import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
-import org.springframework.data.repository.query.ResultProcessor;
-import org.springframework.data.repository.query.ReturnedType;
-import org.springframework.data.util.StreamUtils;
-import org.springframework.util.ClassUtils;
-
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.data.cassandra.mapping.CassandraMappingContext;
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.repository.query.ResultProcessor;
+import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.util.ClassUtils;
+
 /**
  * Reactive query executions for Cassandra.
  *
@@ -78,7 +76,8 @@ interface ReactiveCassandraQueryExecution {
 	}
 
 	/**
-	 * An {@link ReactiveCassandraQueryExecution} that wraps the results of the given delegate with the given result processing.
+	 * An {@link ReactiveCassandraQueryExecution} that wraps the results of the given delegate with the given result
+	 * processing.
 	 *
 	 * @author Mark Paluch
 	 */
@@ -106,6 +105,8 @@ interface ReactiveCassandraQueryExecution {
 	final class ResultProcessingConverter implements Converter<Object, Object> {
 
 		private final @NonNull ResultProcessor processor;
+		private final @NonNull CassandraMappingContext mappingContext;
+		private final @NonNull EntityInstantiators instantiators;
 
 		/* (non-Javadoc)
 		 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
@@ -119,7 +120,14 @@ interface ReactiveCassandraQueryExecution {
 				return source;
 			}
 
-			return processor.processResult(source);
+			if (source != null && returnedType.isInstance(source)) {
+				return source;
+			}
+
+			Converter<Object, Object> converter = new DtoInstantiatingConverter(returnedType.getReturnedType(),
+					mappingContext, instantiators);
+
+			return processor.processResult(source, converter);
 		}
 	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java

@@ -17,6 +17,12 @@ package org.springframework.data.cassandra.repository;
 
 import static org.assertj.core.api.Assertions.*;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.TestSubscriber;
+import rx.Observable;
+import rx.Single;
+
 import java.util.Arrays;
 
 import org.junit.Before;
@@ -41,11 +47,6 @@ import com.datastax.driver.core.KeyspaceMetadata;
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.TableMetadata;
 
-import reactor.core.publisher.Mono;
-import reactor.test.TestSubscriber;
-import rx.Observable;
-import rx.Single;
-
 /**
  * Test for {@link ReactiveCassandraRepository} using reactive wrapper type conversion.
  *
@@ -123,6 +124,22 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 		subscriber.awaitAndAssertNextValueCount(1).assertValues(boyd);
 	}
 
+	/**
+	 * @see DATACASS-360
+	 */
+	@Test
+	public void dtoProjectionShouldWork() {
+
+		TestSubscriber<PersonDto> subscriber = TestSubscriber
+				.subscribe(reactivePersonRepostitory.findProjectedByLastname(boyd.getLastname()));
+
+		subscriber.awaitAndAssertNextValueCount(1).assertValuesWith(personDto -> {
+
+			assertThat(personDto.firstname).isEqualTo(boyd.getFirstname());
+			assertThat(personDto.lastname).isEqualTo(boyd.getLastname());
+		});
+	}
+
 	/**
 	 * @see DATACASS-335
 	 */
@@ -226,6 +243,8 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	interface PersonRepostitory extends ReactiveCrudRepository<Person, String> {
 
 		Publisher<Person> findByLastname(String lastname);
+
+		Flux<PersonDto> findProjectedByLastname(String lastname);
 	}
 
 	@Repository
@@ -252,4 +271,15 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 
 		String getFirstname();
 	}
+
+	static class PersonDto {
+
+		public String firstname, lastname;
+
+		public PersonDto(String firstname, String lastname) {
+
+			this.firstname = firstname;
+			this.lastname = lastname;
+		}
+	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/repository/querymethods/derived/QueryDerivationIntegrationTests.java

@@ -23,8 +23,6 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
-import com.datastax.driver.core.Session;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -47,6 +45,8 @@ import org.springframework.data.util.Version;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import com.datastax.driver.core.Session;
+
 /**
  * Integration tests for query derivation through {@link PersonRepository}.
  *