diff --git a/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt b/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt
index 57b82a88f7..8f60efa9a4 100644
--- a/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt
+++ b/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt
@@ -22,8 +22,9 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.reactive.awaitFirstOrNull
 import kotlinx.coroutines.reactive.awaitSingle
 import kotlinx.coroutines.reactive.flow.asFlow
-import kotlinx.coroutines.reactive.flow.asPublisher
 import org.springframework.core.ParameterizedTypeReference
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import java.net.URI
 
 /**
@@ -53,16 +54,18 @@ suspend fun RSocketRequester.Builder.connectTcpAndAwait(host: String, port: Int)
 suspend fun RSocketRequester.Builder.connectWebSocketAndAwait(uri: URI): RSocketRequester =
 		connectWebSocket(uri).awaitSingle()
 
-
 /**
- * Kotlin [Flow] variant of [RSocketRequester.RequestSpec.data].
+ * Extension for [RSocketRequester.RequestSpec.data] providing a `data<Foo>(producer)`
+ * variant leveraging Kotlin reified type parameters. This extension is not subject to type
+ * erasure and retains actual generic type arguments.
  *
  * @author Sebastien Deleuze
  * @since 5.2
  */
+@Suppress("EXTENSION_SHADOWED_BY_MEMBER")
 @FlowPreview
-fun <T : Any> RSocketRequester.RequestSpec.dataFlow(data: Flow<T>): RSocketRequester.ResponseSpec =
-		data(data.asPublisher(), object : ParameterizedTypeReference<T>() {})
+fun <T : Any> RSocketRequester.RequestSpec.data(producer: Any): RSocketRequester.ResponseSpec =
+		data(producer, object : ParameterizedTypeReference<T>() {})
 
 /**
  * Coroutines variant of [RSocketRequester.ResponseSpec.send].
@@ -92,3 +95,26 @@ suspend fun <T : Any> RSocketRequester.ResponseSpec.retrieveAndAwait(): T =
 @FlowPreview
 fun <T : Any> RSocketRequester.ResponseSpec.retrieveFlow(batchSize: Int = 1): Flow<T> =
 		retrieveFlux(object : ParameterizedTypeReference<T>() {}).asFlow(batchSize)
+
+/**
+ * Extension for [RSocketRequester.ResponseSpec.retrieveMono] providing a `retrieveMono<Foo>()`
+ * variant leveraging Kotlin reified type parameters. This extension is not subject to type
+ * erasure and retains actual generic type arguments.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.2
+ */
+fun <T : Any> RSocketRequester.ResponseSpec.retrieveMono(): Mono<T> =
+		retrieveMono(object : ParameterizedTypeReference<T>() {})
+
+
+/**
+ * Extension for [RSocketRequester.ResponseSpec.retrieveFlux] providing a `retrieveFlux<Foo>()`
+ * variant leveraging Kotlin reified type parameters. This extension is not subject to type
+ * erasure and retains actual generic type arguments.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.2
+ */
+fun <T : Any> RSocketRequester.ResponseSpec.retrieveFlux(): Flux<T> =
+		retrieveFlux(object : ParameterizedTypeReference<T>() {})
diff --git a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt
index 9bb7f1a0f6..9750734a8b 100644
--- a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt
+++ b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt
@@ -3,7 +3,6 @@ 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
@@ -54,11 +53,19 @@ class RSocketRequesterExtensionsTests {
 	}
 
 	@Test
-	fun dataFlow() {
+	fun dataFlowWithType() {
 		val requestSpec = mockk<RSocketRequester.RequestSpec>()
 		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
 		every { requestSpec.data(any<Publisher<String>>(), any<ParameterizedTypeReference<String>>()) } returns responseSpec
-		assertEquals(responseSpec, requestSpec.dataFlow(mockk<Flow<String>>()))
+		assertEquals(responseSpec, requestSpec.data<String>(mockk()))
+	}
+
+	@Test
+	fun dataFlowWithoutType() {
+		val requestSpec = mockk<RSocketRequester.RequestSpec>()
+		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
+		every { requestSpec.data(any()) } returns responseSpec
+		assertEquals(responseSpec, requestSpec.data(mockk()))
 	}
 
 	@Test
@@ -88,4 +95,22 @@ class RSocketRequesterExtensionsTests {
 			assertEquals(listOf("foo", "bar"), responseSpec.retrieveFlow<String>().toList())
 		}
 	}
+
+	@Test
+	fun retrieveMono() {
+		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
+		every { responseSpec.retrieveMono(any<ParameterizedTypeReference<String>>()) } returns Mono.just("foo")
+		runBlocking {
+			assertEquals("foo", responseSpec.retrieveMono<String>().block())
+		}
+	}
+
+	@Test
+	fun retrieveFlux() {
+		val responseSpec = mockk<RSocketRequester.ResponseSpec>()
+		every { responseSpec.retrieveFlux(any<ParameterizedTypeReference<String>>()) } returns Flux.just("foo", "bar")
+		runBlocking {
+			assertEquals(listOf("foo", "bar"), responseSpec.retrieveFlux<String>().collectList().block())
+		}
+	}
 }