Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve RSocketRequester.ResponseSpec Kotlin extensions

Browse Source 
This commit adds retrieveMono and retrieveFlux reified variants, and
turns dataFlow(flow: Flow) extension into a general purpose reified
data(producer: Any) one.

Closes gh-23164
This commit is contained in:
Sebastien Deleuze
2019-06-21 14:44:15 +02:00
parent 003247dc40
commit eaecf0c006
2 changed files with 59 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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>() {})

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

@@ -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())
}
}
}