Adopt to Reactor 2022 changes.

Closes: #1285

Author: christophstrobl

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java

@@ -18,14 +18,12 @@
 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 @@
  *
  * @author Mark Paluch
  * @author Stephen Cohen
+ * @author Christoph Strobl
  */
 public abstract class AbstractR2dbcQuery implements RepositoryQuery {
 
@@ -83,13 +82,12 @@ public R2dbcQueryMethod getQueryMethod() {
 	 */
 	public Object execute(Object[] parameters) {
 
-		RelationalParameterAccessor parameterAccessor = new RelationalParametersParameterAccessor(method, parameters);
-
-		return createQuery(parameterAccessor).flatMapMany(it -> executeQuery(parameterAccessor, it));
+		Mono<R2dbcParameterAccessor> 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);
 

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 @@
  * 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<MonoProcessor<?>> subscriptions;
+	private final R2dbcQueryMethod method;
 
 	/**
 	 * Creates a new {@link R2dbcParameterAccessor}.
@@ -45,37 +49,7 @@ public R2dbcParameterAccessor(R2dbcQueryMethod method, Object... values) {
 		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> 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 @@ public Object[] getValues() {
 	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<R2dbcParameterAccessor> 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<Integer, Optional<?>> holder = new ConcurrentHashMap<>();
+		List<Publisher<?>> 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);
+		});
+	}
 }

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/query/PartTreeR2dbcQueryUnitTests.java

@@ -734,6 +734,20 @@ void createQueryWithPessimisticReadLock() throws Exception {
 				.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 @@ interface UserRepository extends Repository<User, Long> {
 		Mono<Integer> deleteByFirstName(String firstName);
 
 		Mono<Long> countByFirstName(String firstName);
+
+		Mono<User> findByFirstName(Mono<String> firstName);
+
 	}
 
 	@Table("users")