Merge 255ef569d7 into 7e6874ad80

Closes gh-35580

framework-docs/modules/ROOT/pages/core/beans/annotation-config/autowired.adoc

@@ -37,18 +37,18 @@ Kotlin::
 ----
 ======
 
-[NOTE]
+[TIP]
 ====
-As of Spring Framework 4.3, an `@Autowired` annotation on such a constructor is no longer
-necessary if the target bean defines only one constructor to begin with. However, if
-several constructors are available and there is no primary/default constructor, at least
-one of the constructors must be annotated with `@Autowired` in order to instruct the
-container which one to use. See the discussion on
-xref:core/beans/annotation-config/autowired.adoc#beans-autowired-annotation-constructor-resolution[constructor resolution] for details.
+An `@Autowired` annotation on such a constructor is not necessary if the target bean
+defines only one constructor. However, if several constructors are available and there is
+no primary or default constructor, at least one of the constructors must be annotated
+with `@Autowired` in order to instruct the container which one to use. See the discussion
+on xref:core/beans/annotation-config/autowired.adoc#beans-autowired-annotation-constructor-resolution[constructor resolution]
+for details.
 ====
 
-You can also apply the `@Autowired` annotation to _traditional_ setter methods,
-as the following example shows:
+You can apply the `@Autowired` annotation to _traditional_ setter methods, as the
+following example shows:
 
 [tabs]
 ======
@@ -84,8 +84,8 @@ Kotlin::
 ----
 ======
 
-You can also apply the annotation to methods with arbitrary names and multiple
-arguments, as the following example shows:
+You can apply `@Autowired` to methods with arbitrary names and multiple arguments, as the
+following example shows:
 
 [tabs]
 ======
@@ -176,14 +176,15 @@ Kotlin::
 ====
 Make sure that your target components (for example, `MovieCatalog` or `CustomerPreferenceDao`)
 are consistently declared by the type that you use for your `@Autowired`-annotated
-injection points. Otherwise, injection may fail due to a "no type match found" error at runtime.
+injection points. Otherwise, injection may fail due to a "no type match found" error at
+runtime.
 
 For XML-defined beans or component classes found via classpath scanning, the container
 usually knows the concrete type up front. However, for `@Bean` factory methods, you need
 to make sure that the declared return type is sufficiently expressive. For components
 that implement several interfaces or for components potentially referred to by their
-implementation type, consider declaring the most specific return type on your factory
-method (at least as specific as required by the injection points referring to your bean).
+implementation type, declare the most specific return type on your factory method (at
+least as specific as required by the injection points referring to your bean).
 ====
 
 .[[beans-autowired-annotation-self-injection]]Self Injection
@@ -312,8 +313,8 @@ through `@Order` values in combination with `@Primary` on a single bean for each
 ====
 
 Even typed `Map` instances can be autowired as long as the expected key type is `String`.
-The map values contain all beans of the expected type, and the keys contain the
-corresponding bean names, as the following example shows:
+The map values are all beans of the expected type, and the keys are the corresponding
+bean names, as the following example shows:
 
 [tabs]
 ======
@@ -431,7 +432,7 @@ annotated constructor does not have to be public.
 ====
 
 Alternatively, you can express the non-required nature of a particular dependency
-through Java 8's `java.util.Optional`, as the following example shows:
+through Java's `java.util.Optional`, as the following example shows:
 
 [source,java,indent=0,subs="verbatim,quotes"]
 ----
@@ -445,8 +446,8 @@ through Java 8's `java.util.Optional`, as the following example shows:
 ----
 
 You can also use a parameter-level `@Nullable` annotation (of any kind in any package --
-for example, `javax.annotation.Nullable` from JSR-305) or just leverage Kotlin built-in
-null-safety support:
+for example, `org.jspecify.annotations.Nullable` from JSpecify) or just leverage Kotlin's
+built-in null-safety support:
 
 [tabs]
 ======
@@ -477,13 +478,6 @@ Kotlin::
 ----
 ======
 
-[NOTE]
-====
-A type-level `@Nullable` annotation such as from JSpecify is not supported in Spring
-Framework 6.2 yet. You need to upgrade to Spring Framework 7.0 where the framework
-detects type-level annotations and consistently declares JSpecify in its own codebase.
-====
-
 You can also use `@Autowired` for interfaces that are well-known resolvable
 dependencies: `BeanFactory`, `ApplicationContext`, `Environment`, `ResourceLoader`,
 `ApplicationEventPublisher`, and `MessageSource`. These interfaces and their extended
@@ -528,5 +522,6 @@ class MovieRecommender {
 The `@Autowired`, `@Inject`, `@Value`, and `@Resource` annotations are handled by Spring
 `BeanPostProcessor` implementations. This means that you cannot apply these annotations
 within your own `BeanPostProcessor` or `BeanFactoryPostProcessor` types (if any).
+
 These types must be 'wired up' explicitly by using XML or a Spring `@Bean` method.
 ====

spring-aop/src/main/java/org/springframework/aop/framework/CoroutinesUtils.java

@@ -17,6 +17,7 @@
 package org.springframework.aop.framework;
 
 import kotlin.coroutines.Continuation;
+import kotlinx.coroutines.flow.Flow;
 import kotlinx.coroutines.reactive.ReactiveFlowKt;
 import kotlinx.coroutines.reactor.MonoKt;
 import org.jspecify.annotations.Nullable;
@@ -35,6 +36,9 @@ abstract class CoroutinesUtils {
 		if (publisher instanceof Publisher<?> rsPublisher) {
 			return ReactiveFlowKt.asFlow(rsPublisher);
 		}
+		else if (publisher instanceof Flow<?>) {
+			return publisher;
+		}
 		else {
 			throw new IllegalArgumentException("Not a Reactive Streams Publisher: " + publisher);
 		}

spring-aop/src/test/kotlin/org/springframework/aop/framework/CoroutinesUtilsTests.kt

@@ -18,6 +18,7 @@ package org.springframework.aop.framework
 
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
@@ -72,4 +73,16 @@ class CoroutinesUtilsTests {
 		}
 	}
 
+	@Test
+	@Suppress("UNCHECKED_CAST")
+	fun flowAsFlow() {
+		val value1 = "foo"
+		val value2 = "bar"
+		val values = flowOf(value1, value2)
+		val flow = CoroutinesUtils.asFlow(values) as Flow<String>
+		runBlocking {
+			assertThat(flow.toList()).containsExactly(value1, value2)
+		}
+	}
+
 }

spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceMethod.java

@@ -30,6 +30,7 @@ import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.DefaultParameterNameDiscoverer;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ReactiveAdapter;
@@ -54,6 +55,8 @@ import org.springframework.util.StringValueResolver;
  */
 final class RSocketServiceMethod {
 
+	private static final String COROUTINES_FLOW_CLASS_NAME = "kotlinx.coroutines.flow.Flow";
+
 	private final Method method;
 
 	private final MethodParameter[] parameters;
@@ -82,6 +85,10 @@ final class RSocketServiceMethod {
 		if (count == 0) {
 			return new MethodParameter[0];
 		}
+		if (KotlinDetector.isSuspendingFunction(method)) {
+			count -= 1;
+		}
+
 		DefaultParameterNameDiscoverer nameDiscoverer = new DefaultParameterNameDiscoverer();
 		MethodParameter[] parameters = new MethodParameter[count];
 		for (int i = 0; i < count; i++) {
@@ -129,10 +136,16 @@ final class RSocketServiceMethod {
 
 		MethodParameter returnParam = new MethodParameter(method, -1);
 		Class<?> returnType = returnParam.getParameterType();
+		boolean isUnwrapped = KotlinDetector.isSuspendingFunction(method) &&
+				!COROUTINES_FLOW_CLASS_NAME.equals(returnParam.getParameterType().getName());
+		if (isUnwrapped) {
+			returnType = Mono.class;
+		}
+
 		ReactiveAdapter reactiveAdapter = reactiveRegistry.getAdapter(returnType);
 
 		MethodParameter actualParam = (reactiveAdapter != null ? returnParam.nested() : returnParam.nestedIfOptional());
-		Class<?> actualType = actualParam.getNestedParameterType();
+		Class<?> actualType = isUnwrapped ? actualParam.getParameterType() : actualParam.getNestedParameterType();
 
 		Function<RSocketRequestValues, Publisher<?>> responseFunction;
 		if (ClassUtils.isVoidType(actualType) || (reactiveAdapter != null && reactiveAdapter.isNoValue())) {
@@ -147,7 +160,8 @@ final class RSocketServiceMethod {
 		}
 		else {
 			ParameterizedTypeReference<?> payloadType =
-					ParameterizedTypeReference.forType(actualParam.getNestedGenericParameterType());
+					ParameterizedTypeReference.forType(isUnwrapped ? actualParam.getGenericParameterType() :
+							actualParam.getNestedGenericParameterType());
 
 			responseFunction = values -> (
 					reactiveAdapter.isMultiValue() ?

spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceProxyFactory.java

@@ -31,6 +31,7 @@ import org.jspecify.annotations.Nullable;
 
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.framework.ReflectiveMethodInvocation;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.MethodIntrospector;
 import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.annotation.AnnotatedElementUtils;
@@ -246,7 +247,9 @@ public final class RSocketServiceProxyFactory {
 			Method method = invocation.getMethod();
 			RSocketServiceMethod serviceMethod = this.serviceMethods.get(method);
 			if (serviceMethod != null) {
-				return serviceMethod.invoke(invocation.getArguments());
+				@Nullable Object[] arguments = KotlinDetector.isSuspendingFunction(method) ?
+						resolveCoroutinesArguments(invocation.getArguments()) : invocation.getArguments();
+				return serviceMethod.invoke(arguments);
 			}
 			if (method.isDefault()) {
 				if (invocation instanceof ReflectiveMethodInvocation reflectiveMethodInvocation) {
@@ -256,6 +259,12 @@ public final class RSocketServiceProxyFactory {
 			}
 			throw new IllegalStateException("Unexpected method invocation: " + method);
 		}
+
+		private static Object[] resolveCoroutinesArguments(@Nullable Object[] args) {
+			Object[] functionArgs = new Object[args.length - 1];
+			System.arraycopy(args, 0, functionArgs, 0, args.length - 1);
+			return functionArgs;
+		}
 	}
 
 }

spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/service/RSocketServiceMethodKotlinTests.kt

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2002-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.rsocket.service
+
+import io.rsocket.util.DefaultPayload
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.reactive.asFlow
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.springframework.messaging.rsocket.RSocketRequester
+import org.springframework.messaging.rsocket.RSocketStrategies
+import org.springframework.messaging.rsocket.TestRSocket
+import org.springframework.util.MimeTypeUtils.TEXT_PLAIN
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * Kotlin tests for [RSocketServiceMethod].
+ *
+ * @author Dmitry Sulman
+ */
+class RSocketServiceMethodKotlinTests {
+
+	private lateinit var rsocket: TestRSocket
+
+	private lateinit var proxyFactory: RSocketServiceProxyFactory
+
+	@BeforeEach
+	fun setUp() {
+		rsocket = TestRSocket()
+		val requester = RSocketRequester.wrap(rsocket, TEXT_PLAIN, TEXT_PLAIN, RSocketStrategies.create())
+		proxyFactory = RSocketServiceProxyFactory.builder(requester).build()
+	}
+
+	@Test
+	fun fireAndForget(): Unit = runBlocking {
+		val service = proxyFactory.createClient(SuspendingFunctionsService::class.java)
+
+		val requestPayload = "request"
+		service.fireAndForget(requestPayload)
+
+		assertThat(rsocket.savedMethodName).isEqualTo("fireAndForget")
+		assertThat(rsocket.savedPayload?.metadataUtf8).isEqualTo("ff")
+		assertThat(rsocket.savedPayload?.dataUtf8).isEqualTo(requestPayload)
+	}
+
+	@Test
+	fun requestResponse(): Unit = runBlocking {
+		val service = proxyFactory.createClient(SuspendingFunctionsService::class.java)
+
+		val requestPayload = "request"
+		val responsePayload = "response"
+		rsocket.setPayloadMonoToReturn(Mono.just(DefaultPayload.create(responsePayload)))
+		val response = service.requestResponse(requestPayload)
+
+		assertThat(response).isEqualTo(responsePayload)
+		assertThat(rsocket.savedMethodName).isEqualTo("requestResponse")
+		assertThat(rsocket.savedPayload?.metadataUtf8).isEqualTo("rr")
+		assertThat(rsocket.savedPayload?.dataUtf8).isEqualTo(requestPayload)
+	}
+
+	@Test
+	fun requestStream(): Unit = runBlocking {
+		val service = proxyFactory.createClient(SuspendingFunctionsService::class.java)
+
+		val requestPayload = "request"
+		val responsePayload1 = "response1"
+		val responsePayload2 = "response2"
+		rsocket.setPayloadFluxToReturn(
+			Flux.just(DefaultPayload.create(responsePayload1), DefaultPayload.create(responsePayload2)))
+		val response = service.requestStream(requestPayload).toList()
+
+		assertThat(response).containsExactly(responsePayload1, responsePayload2)
+		assertThat(rsocket.savedMethodName).isEqualTo("requestStream")
+		assertThat(rsocket.savedPayload?.metadataUtf8).isEqualTo("rs")
+		assertThat(rsocket.savedPayload?.dataUtf8).isEqualTo(requestPayload)
+	}
+
+	@Test
+	fun requestChannel(): Unit = runBlocking {
+		val service = proxyFactory.createClient(SuspendingFunctionsService::class.java)
+
+		val requestPayload1 = "request1"
+		val requestPayload2 = "request2"
+		val responsePayload1 = "response1"
+		val responsePayload2 = "response2"
+		rsocket.setPayloadFluxToReturn(
+			Flux.just(DefaultPayload.create(responsePayload1), DefaultPayload.create(responsePayload2)))
+		val response = service.requestChannel(flowOf(requestPayload1, requestPayload2)).toList()
+
+		assertThat(response).containsExactly(responsePayload1, responsePayload2)
+		assertThat(rsocket.savedMethodName).isEqualTo("requestChannel")
+
+		val savedPayloads = rsocket.savedPayloadFlux
+			?.asFlow()
+			?.map { it.dataUtf8 }
+			?.toList()
+		assertThat(savedPayloads).containsExactly(requestPayload1, requestPayload2)
+	}
+
+	private interface SuspendingFunctionsService {
+
+		@RSocketExchange("ff")
+		suspend fun fireAndForget(input: String)
+
+		@RSocketExchange("rr")
+		suspend fun requestResponse(input: String): String
+
+		@RSocketExchange("rs")
+		suspend fun requestStream(input: String): Flow<String>
+
+		@RSocketExchange("rc")
+		suspend fun requestChannel(input: Flow<String>): Flow<String>
+	}
+}