Avoid shadowed RSocketRequester extensions

This commit renames RSocketRequester.RequestSpec data extension to
dataWithType to avoid shadowing issues and adds Publisher<T> and Flow<T>
variants to provide automatic reified type parameter resolution for
those types. It also makes RSocketRequester consistent with the
changes introduced via 2b4d6ce354 in WebFlux.

spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt

@@ -22,6 +22,7 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.reactive.awaitFirstOrNull
 import kotlinx.coroutines.reactive.awaitSingle
 import kotlinx.coroutines.reactive.flow.asFlow
+import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
@@ -55,18 +56,45 @@ suspend fun RSocketRequester.Builder.connectWebSocketAndAwait(uri: URI): RSocket
 		connectWebSocket(uri).awaitSingle()
 
 /**
- * Extension for [RSocketRequester.RequestSpec.data] providing a `data<Foo>(producer)`
+ * Extension for [RSocketRequester.RequestSpec.data] providing a `dataWithType<Foo>(Any)`
  * variant leveraging Kotlin reified type parameters. This extension is not subject to type
  * erasure and retains actual generic type arguments.
- *
+ * @param producer the source of payload data value(s). This must be a
+ * [Publisher] or another producer adaptable to a
+ * [Publisher] via [org.springframework.core.ReactiveAdapterRegistry]
+ * @param <T> the type of values to be produced
  * @author Sebastien Deleuze
  * @since 5.2
  */
-@Suppress("EXTENSION_SHADOWED_BY_MEMBER")
-@FlowPreview
-inline fun <reified T : Any> RSocketRequester.RequestSpec.data(producer: Any): RSocketRequester.ResponseSpec =
+inline fun <reified T : Any> RSocketRequester.RequestSpec.dataWithType(producer: Any): RSocketRequester.ResponseSpec =
 		data(producer, object : ParameterizedTypeReference<T>() {})
 
+/**
+ * Extension for [RSocketRequester.RequestSpec.data] providing a `dataWithType(Publisher<T>)`
+ * variant leveraging Kotlin reified type parameters. This extension is not subject to type
+ * erasure and retains actual generic type arguments.
+ * @param publisher the source of payload data value(s)
+ * @param <T> the type of values to be produced
+ * @author Sebastien Deleuze
+ * @since 5.2
+ */
+inline fun <reified T : Any> RSocketRequester.RequestSpec.dataWithType(publisher: Publisher<T>): RSocketRequester.ResponseSpec =
+		data(publisher, object : ParameterizedTypeReference<T>() {})
+
+/**
+ * Extension for [RSocketRequester.RequestSpec.data] providing a `dataWithType(Flow<T>)`
+ * variant leveraging Kotlin reified type parameters. This extension is not subject to type
+ * erasure and retains actual generic type arguments.
+ * @param flow the [Flow] to write to the request
+ * @param <T> the source of payload data value(s)
+ * @author Sebastien Deleuze
+ * @since 5.2
+ */
+@FlowPreview
+inline fun <reified T : Any> RSocketRequester.RequestSpec.dataWithType(flow: Flow<T>): RSocketRequester.ResponseSpec =
+		data(flow, object : ParameterizedTypeReference<T>() {})
+
+
 /**
  * Coroutines variant of [RSocketRequester.ResponseSpec.send].
  *

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

@@ -3,6 +3,7 @@ package org.springframework.messaging.rsocket
 import io.mockk.every
 import io.mockk.mockk
 import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
@@ -12,6 +13,7 @@ import org.reactivestreams.Publisher
 import org.springframework.core.ParameterizedTypeReference
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.util.concurrent.CompletableFuture
 
 /**
  * Mock object based tests for [RSocketRequester] Kotlin extensions
@@ -55,11 +57,30 @@ class RSocketRequesterExtensionsTests {
 	}
 
 	@Test
-	fun dataFlowWithType() {
+	fun `dataWithType with Publisher`() {
 		val requestSpec = mockk<RSocketRequester.RequestSpec>()
 		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
+		val data = mockk<Publisher<String>>()
 		every { requestSpec.data(any<Publisher<String>>(), match<ParameterizedTypeReference<*>>(stringTypeRefMatcher)) } returns responseSpec
-		assertEquals(responseSpec, requestSpec.data<String>(mockk()))
+		assertEquals(responseSpec, requestSpec.dataWithType(data))
+	}
+
+	@Test
+	fun `dataWithType with Flow`() {
+		val requestSpec = mockk<RSocketRequester.RequestSpec>()
+		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
+		val data = mockk<Flow<String>>()
+		every { requestSpec.data(any<Publisher<String>>(), match<ParameterizedTypeReference<*>>(stringTypeRefMatcher)) } returns responseSpec
+		assertEquals(responseSpec, requestSpec.dataWithType(data))
+	}
+
+	@Test
+	fun `dataWithType with CompletableFuture`() {
+		val requestSpec = mockk<RSocketRequester.RequestSpec>()
+		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
+		val data = mockk<CompletableFuture<String>>()
+		every { requestSpec.data(any<Publisher<String>>(), match<ParameterizedTypeReference<*>>(stringTypeRefMatcher)) } returns responseSpec
+		assertEquals(responseSpec, requestSpec.dataWithType<String>(data))
 	}
 
 	@Test