diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java index 9834e6ae0..5a70ed45f 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java +++ b/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 resultProcessing) { - return new ResultProcessingExecution(getExecutionToWrap(accessor, resultProcessing), resultProcessing); + return new ResultProcessingExecution(getExecutionToWrap(), resultProcessing); } - private ReactiveCassandraQueryExecution getExecutionToWrap(CassandraParameterAccessor accessor, - Converter resultProcessing) { + private ReactiveCassandraQueryExecution getExecutionToWrap() { - return (method.isCollectionQuery() ? new CollectionExecution(operations) - : new SingleEntityExecution(operations)); + return (method.isCollectionQuery() ? new CollectionExecution(operations) : new SingleEntityExecution(operations)); } /** diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java index d6b08b878..2138ce535 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java +++ b/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 { 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 converter = new DtoInstantiatingConverter(returnedType.getReturnedType(), + mappingContext, instantiators); + + return processor.processResult(source, converter); } } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java index cdfff9865..6a2740b99 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java +++ b/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 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 { Publisher findByLastname(String lastname); + + Flux 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; + } + } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/repository/querymethods/derived/QueryDerivationIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/repository/querymethods/derived/QueryDerivationIntegrationTests.java index 8fb9d469d..0c71bfef1 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/repository/querymethods/derived/QueryDerivationIntegrationTests.java +++ b/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}. *