From dd3a69c20d6d6a7dc05fcc67587e8763d18840cd Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Tue, 5 Jul 2022 09:27:18 +0200 Subject: [PATCH] Adopt to Reactor 2022 changes. Closes: #1285 --- .../repository/query/AbstractR2dbcQuery.java | 10 +- .../query/R2dbcParameterAccessor.java | 97 ++++++++++++------- .../query/PartTreeR2dbcQueryUnitTests.java | 17 ++++ 3 files changed, 85 insertions(+), 39 deletions(-) diff --git a/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java b/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java index 18b1d229..d0e21f92 100644 --- a/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java +++ b/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java @@ -18,14 +18,12 @@ package org.springframework.data.r2dbc.repository.query; import reactor.core.publisher.Mono; import org.reactivestreams.Publisher; - import org.springframework.data.mapping.model.EntityInstantiators; import org.springframework.data.r2dbc.convert.R2dbcConverter; import org.springframework.data.r2dbc.core.R2dbcEntityOperations; import org.springframework.data.r2dbc.repository.query.R2dbcQueryExecution.ResultProcessingConverter; import org.springframework.data.r2dbc.repository.query.R2dbcQueryExecution.ResultProcessingExecution; import org.springframework.data.relational.repository.query.RelationalParameterAccessor; -import org.springframework.data.relational.repository.query.RelationalParametersParameterAccessor; import org.springframework.data.repository.query.ParameterAccessor; import org.springframework.data.repository.query.RepositoryQuery; import org.springframework.data.repository.query.ResultProcessor; @@ -41,6 +39,7 @@ import org.springframework.util.Assert; * * @author Mark Paluch * @author Stephen Cohen + * @author Christoph Strobl */ public abstract class AbstractR2dbcQuery implements RepositoryQuery { @@ -83,13 +82,12 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery { */ public Object execute(Object[] parameters) { - RelationalParameterAccessor parameterAccessor = new RelationalParametersParameterAccessor(method, parameters); - - return createQuery(parameterAccessor).flatMapMany(it -> executeQuery(parameterAccessor, it)); + Mono resolveParameters = new R2dbcParameterAccessor(method, parameters).resolveParameters(); + return resolveParameters.flatMapMany(it -> createQuery(it).flatMapMany(foo -> executeQuery(it, foo))); } @SuppressWarnings("unchecked") - private Publisher executeQuery(RelationalParameterAccessor parameterAccessor, PreparedOperation operation) { + private Publisher executeQuery(R2dbcParameterAccessor parameterAccessor, PreparedOperation operation) { ResultProcessor processor = method.getResultProcessor().withDynamicProjection(parameterAccessor); diff --git a/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcParameterAccessor.java b/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcParameterAccessor.java index 6145f5dd..efb2f909 100644 --- a/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcParameterAccessor.java +++ b/spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcParameterAccessor.java @@ -15,12 +15,15 @@ */ package org.springframework.data.r2dbc.repository.query; +import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoProcessor; import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; import org.springframework.data.relational.repository.query.RelationalParametersParameterAccessor; import org.springframework.data.repository.util.ReactiveWrapperConverters; @@ -31,11 +34,12 @@ import org.springframework.data.repository.util.ReactiveWrappers; * to reactive parameter wrapper types upon creation. This class performs synchronization when accessing parameters. * * @author Mark Paluch + * @author Christoph Strobl */ class R2dbcParameterAccessor extends RelationalParametersParameterAccessor { private final Object[] values; - private final List> subscriptions; + private final R2dbcQueryMethod method; /** * Creates a new {@link R2dbcParameterAccessor}. @@ -45,37 +49,7 @@ class R2dbcParameterAccessor extends RelationalParametersParameterAccessor { super(method, values); this.values = values; - this.subscriptions = new ArrayList<>(values.length); - - for (int i = 0; i < values.length; i++) { - - Object value = values[i]; - - if (value == null || !ReactiveWrappers.supports(value.getClass())) { - subscriptions.add(null); - continue; - } - - if (ReactiveWrappers.isSingleValueType(value.getClass())) { - subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Mono.class).toProcessor()); - } else { - subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().toProcessor()); - } - } - } - - /* (non-Javadoc) - * @see org.springframework.data.repository.query.ParametersParameterAccessor#getValue(int) - */ - @SuppressWarnings("unchecked") - @Override - protected T getValue(int index) { - - if (subscriptions.get(index) != null) { - return (T) subscriptions.get(index).block(); - } - - return super.getValue(index); + this.method = method; } /* (non-Javadoc) @@ -97,4 +71,61 @@ class R2dbcParameterAccessor extends RelationalParametersParameterAccessor { public Object getBindableValue(int index) { return getValue(getParameters().getBindableParameter(index).getIndex()); } + + /** + * Resolve parameters that were provided through reactive wrapper types. Flux is collected into a list, values from + * Mono's are used directly. + * + * @return + */ + @SuppressWarnings("unchecked") + public Mono resolveParameters() { + + boolean hasReactiveWrapper = false; + + for (Object value : values) { + if (value == null || !ReactiveWrappers.supports(value.getClass())) { + continue; + } + + hasReactiveWrapper = true; + break; + } + + if (!hasReactiveWrapper) { + return Mono.just(this); + } + + Object[] resolved = new Object[values.length]; + Map> holder = new ConcurrentHashMap<>(); + List> publishers = new ArrayList<>(); + + for (int i = 0; i < values.length; i++) { + + Object value = resolved[i] = values[i]; + if (value == null || !ReactiveWrappers.supports(value.getClass())) { + continue; + } + + if (ReactiveWrappers.isSingleValueType(value.getClass())) { + + int index = i; + publishers.add(ReactiveWrapperConverters.toWrapper(value, Mono.class) // + .map(Optional::of) // + .defaultIfEmpty(Optional.empty()) // + .doOnNext(it -> holder.put(index, (Optional) it))); + } else { + + int index = i; + publishers.add(ReactiveWrapperConverters.toWrapper(value, Flux.class) // + .collectList() // + .doOnNext(it -> holder.put(index, Optional.of(it)))); + } + } + + return Flux.merge(publishers).then().thenReturn(resolved).map(values -> { + holder.forEach((index, v) -> values[index] = v.orElse(null)); + return new R2dbcParameterAccessor(method, values); + }); + } } diff --git a/spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQueryUnitTests.java b/spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQueryUnitTests.java index 3abd333d..6a73c1b9 100644 --- a/spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQueryUnitTests.java +++ b/spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQueryUnitTests.java @@ -734,6 +734,20 @@ class PartTreeR2dbcQueryUnitTests { .where("users.first_name = $1 AND (users.age = $2) FOR SHARE OF users"); } + @Test // GH-1285 + void bindsParametersFromPublisher() throws Exception { + + R2dbcQueryMethod queryMethod = getQueryMethod("findByFirstName", Mono.class); + PartTreeR2dbcQuery r2dbcQuery = new PartTreeR2dbcQuery(queryMethod, operations, r2dbcConverter, dataAccessStrategy); + R2dbcParameterAccessor accessor = new R2dbcParameterAccessor(queryMethod, new Object[] { Mono.just("John") }); + + PreparedOperation preparedOperation = createQuery(r2dbcQuery, accessor.resolveParameters().block()); + BindTarget bindTarget = mock(BindTarget.class); + preparedOperation.bindTo(bindTarget); + + verify(bindTarget, times(1)).bind(0, "John"); + } + private PreparedOperation createQuery(R2dbcQueryMethod queryMethod, PartTreeR2dbcQuery r2dbcQuery, Object... parameters) { return createQuery(r2dbcQuery, getAccessor(queryMethod, parameters)); @@ -927,6 +941,9 @@ class PartTreeR2dbcQueryUnitTests { Mono deleteByFirstName(String firstName); Mono countByFirstName(String firstName); + + Mono findByFirstName(Mono firstName); + } @Table("users")