Add RSocketRequester retrieveAndAwaitOrNull extension

Closes gh-23874

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

@@ -112,6 +112,15 @@ suspend fun RSocketRequester.RequestSpec.sendAndAwait() {
 suspend inline fun <reified T : Any> RSocketRequester.RequestSpec.retrieveAndAwait(): T =
 		retrieveMono(object : ParameterizedTypeReference<T>() {}).awaitSingle()
 
+/**
+ * Nullable coroutines variant of [RSocketRequester.RequestSpec.retrieveMono].
+ *
+ * @author Sebastien Deleuze
+ * @since 5.2.1
+ */
+suspend inline fun <reified T : Any> RSocketRequester.RequestSpec.retrieveAndAwaitOrNull(): T? =
+		retrieveMono(object : ParameterizedTypeReference<T>() {}).awaitFirstOrNull()
+
 /**
  * Coroutines variant of [RSocketRequester.RequestSpec.retrieveFlux].
  *

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

@@ -120,6 +120,15 @@ class RSocketRequesterExtensionsTests {
 		}
 	}
 
+	@Test
+	fun retrieveAndAwaitOrNull() {
+		val requestSpec = mockk<RSocketRequester.RequestSpec>()
+		every { requestSpec.retrieveMono(match<ParameterizedTypeReference<*>>(stringTypeRefMatcher)) } returns Mono.empty()
+		runBlocking {
+			assertThat(requestSpec.retrieveAndAwaitOrNull<String>()).isNull()
+		}
+	}
+
 	@Test
 	fun retrieveFlow() {
 		val requestSpec = mockk<RSocketRequester.RequestSpec>()