diff --git a/build.gradle b/build.gradle index e097cf14ce..769c6dd19a 100644 --- a/build.gradle +++ b/build.gradle @@ -29,7 +29,7 @@ ext { } aspectjVersion = "1.9.2" - coroutinesVersion = "1.2.0-alpha" + coroutinesVersion = "1.2.0-alpha-2" freemarkerVersion = "2.3.28" groovyVersion = "2.5.6" hsqldbVersion = "2.4.1" diff --git a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java index 7b9db3c16d..ea88d9497b 100644 --- a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java +++ b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java @@ -27,6 +27,9 @@ import io.reactivex.BackpressureStrategy; import io.reactivex.Flowable; import kotlinx.coroutines.CompletableDeferredKt; import kotlinx.coroutines.Deferred; +import kotlinx.coroutines.flow.FlowKt; +import kotlinx.coroutines.reactive.flow.FlowAsPublisherKt; +import kotlinx.coroutines.reactive.flow.PublisherAsFlowKt; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -97,6 +100,10 @@ public class ReactiveAdapterRegistry { if (ClassUtils.isPresent("kotlinx.coroutines.Deferred", classLoader)) { new CoroutinesRegistrar().registerAdapters(this); } + // TODO Use a single CoroutinesRegistrar when Flow will be not experimental anymore + if (ClassUtils.isPresent("kotlinx.coroutines.flow.Flow", classLoader)) { + new CoroutinesFlowRegistrar().registerAdapters(this); + } } @@ -335,7 +342,17 @@ public class ReactiveAdapterRegistry { source -> CoroutinesUtils.deferredToMono((Deferred) source), source -> CoroutinesUtils.monoToDeferred(Mono.from(source))); } + } + private static class CoroutinesFlowRegistrar { + + void registerAdapters(ReactiveAdapterRegistry registry) { + registry.registerReactiveType( + ReactiveTypeDescriptor.multiValue(kotlinx.coroutines.flow.Flow.class, FlowKt::emptyFlow), + source -> FlowAsPublisherKt.from((kotlinx.coroutines.flow.Flow) source), + PublisherAsFlowKt::from + ); + } } } diff --git a/spring-core/src/test/kotlin/org/springframework/core/KotlinReactiveAdapterRegistryTests.kt b/spring-core/src/test/kotlin/org/springframework/core/KotlinReactiveAdapterRegistryTests.kt index 520138d90e..bbab749d88 100644 --- a/spring-core/src/test/kotlin/org/springframework/core/KotlinReactiveAdapterRegistryTests.kt +++ b/spring-core/src/test/kotlin/org/springframework/core/KotlinReactiveAdapterRegistryTests.kt @@ -17,14 +17,21 @@ package org.springframework.core import kotlinx.coroutines.Deferred +import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.GlobalScope import kotlinx.coroutines.async +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue +import org.junit.Assert.fail import org.junit.Test import org.reactivestreams.Publisher +import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.test.StepVerifier import java.time.Duration import kotlin.reflect.KClass @@ -49,6 +56,36 @@ class KotlinReactiveAdapterRegistryTests { } + @Test + @FlowPreview + fun flowToPublisher() { + val source = flow { + emit(1) + emit(2) + emit(3) + } + val target: Publisher = getAdapter(Flow::class).toPublisher(source) + assertTrue("Expected Flux Publisher: " + target.javaClass.name, target is Flux<*>) + StepVerifier.create(target) + .expectNext(1) + .expectNext(2) + .expectNext(3) + .verifyComplete() + } + + @Test + @FlowPreview + fun publisherToFlow() { + val source = Flux.just(1, 2, 3) + val target = getAdapter(Flow::class).fromPublisher(source) + if (target is Flow<*>) { + assertEquals(listOf(1, 2, 3), runBlocking { target.toList() }) + } + else { + fail() + } + } + private fun getAdapter(reactiveType: KClass<*>): ReactiveAdapter { return this.registry.getAdapter(reactiveType.java)!! } diff --git a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensions.kt b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensions.kt index 4672da062b..3f9e9ed8bc 100644 --- a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensions.kt +++ b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensions.kt @@ -16,8 +16,11 @@ package org.springframework.web.reactive.function.client +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle +import kotlinx.coroutines.reactive.flow.asFlow import org.springframework.core.ParameterizedTypeReference import org.springframework.http.ResponseEntity import reactor.core.publisher.Flux @@ -45,6 +48,19 @@ inline fun ClientResponse.bodyToMono(): Mono = inline fun ClientResponse.bodyToFlux(): Flux = bodyToFlux(object : ParameterizedTypeReference() {}) +/** + * Coroutines [kotlinx.coroutines.flow.Flow] based variant of [ClientResponse.bodyToFlux]. + * + * Backpressure is controlled by [batchSize] parameter that controls the size of in-flight elements + * and [org.reactivestreams.Subscription.request] size. + * + * @author Sebastien Deleuze + * @since 5.2 + */ +@FlowPreview +inline fun ClientResponse.bodyToFlow(batchSize: Int = 1): Flow = + bodyToFlux().asFlow(batchSize) + /** * Extension for [ClientResponse.toEntity] providing a `toEntity()` variant * leveraging Kotlin reified type parameters. This extension is not subject to type diff --git a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt index 67816c74e7..db9e0b8339 100644 --- a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt +++ b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt @@ -17,8 +17,12 @@ package org.springframework.web.reactive.function.client import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.GlobalScope +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.reactive.awaitSingle +import kotlinx.coroutines.reactive.flow.asFlow +import kotlinx.coroutines.reactive.flow.asPublisher import kotlinx.coroutines.reactor.mono import org.reactivestreams.Publisher import org.springframework.core.ParameterizedTypeReference @@ -28,7 +32,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono /** - * Extension for [WebClient.RequestBodySpec.body] providing a `body()` variant + * Extension for [WebClient.RequestBodySpec.body] providing a `body(Publisher)` variant * leveraging Kotlin reified type parameters. This extension is not subject to type * erasure and retains actual generic type arguments. * @@ -39,6 +43,18 @@ import reactor.core.publisher.Mono inline fun > RequestBodySpec.body(publisher: S): RequestHeadersSpec<*> = body(publisher, object : ParameterizedTypeReference() {}) +/** + * Coroutines [Flow] based extension for [WebClient.RequestBodySpec.body] providing a + * body(Flow)` 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 + */ +@FlowPreview +inline fun > RequestBodySpec.body(flow: S): RequestHeadersSpec<*> = + body(flow.asPublisher(), object : ParameterizedTypeReference() {}) + /** * Extension for [WebClient.ResponseSpec.bodyToMono] providing a `bodyToMono()` variant * leveraging Kotlin reified type parameters. This extension is not subject to type @@ -62,6 +78,20 @@ inline fun WebClient.ResponseSpec.bodyToMono(): Mono = inline fun WebClient.ResponseSpec.bodyToFlux(): Flux = bodyToFlux(object : ParameterizedTypeReference() {}) +/** + * Coroutines [kotlinx.coroutines.flow.Flow] based variant of [WebClient.ResponseSpec.bodyToFlux]. + * + * Backpressure is controlled by [batchSize] parameter that controls the size of in-flight elements + * and [org.reactivestreams.Subscription.request] size. + * + * @author Sebastien Deleuze + * @since 5.2 + */ +@FlowPreview +inline fun WebClient.ResponseSpec.bodyToFlow(batchSize: Int = 1): Flow = + bodyToFlux().asFlow(batchSize) + + /** * Coroutines variant of [WebClient.RequestHeadersSpec.exchange]. * diff --git a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensions.kt b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensions.kt index 406cfd1266..e44615da65 100644 --- a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensions.kt +++ b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensions.kt @@ -16,8 +16,11 @@ package org.springframework.web.reactive.function.server +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle +import kotlinx.coroutines.reactive.flow.asFlow import org.springframework.core.ParameterizedTypeReference import org.springframework.http.codec.multipart.Part import org.springframework.util.MultiValueMap @@ -48,6 +51,19 @@ inline fun ServerRequest.bodyToMono(): Mono = inline fun ServerRequest.bodyToFlux(): Flux = bodyToFlux(object : ParameterizedTypeReference() {}) +/** + * Coroutines [kotlinx.coroutines.flow.Flow] based variant of [ServerRequest.bodyToFlux]. + * + * Backpressure is controlled by [batchSize] parameter that controls the size of in-flight elements + * and [org.reactivestreams.Subscription.request] size. + * + * @author Sebastien Deleuze + * @since 5.2 + */ +@FlowPreview +inline fun ServerRequest.bodyToFlow(batchSize: Int = 1): Flow = + bodyToFlux().asFlow(batchSize) + /** * Non-nullable Coroutines variant of [ServerRequest.bodyToMono]. * diff --git a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensions.kt b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensions.kt index 791e25e495..f7cac34324 100644 --- a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensions.kt +++ b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensions.kt @@ -16,7 +16,10 @@ package org.springframework.web.reactive.function.server +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.reactive.awaitSingle +import kotlinx.coroutines.reactive.flow.asPublisher import org.reactivestreams.Publisher import org.springframework.core.ParameterizedTypeReference import org.springframework.http.MediaType @@ -74,6 +77,19 @@ fun ServerResponse.BodyBuilder.html() = contentType(MediaType.TEXT_HTML) suspend fun ServerResponse.HeadersBuilder>.buildAndAwait(): ServerResponse = build().awaitSingle() + +/** + * Coroutines [Flow] based extension for [ServerResponse.BodyBuilder.body] providing a + * `body(Flow)` variant. This extension is not subject to type erasure and retains + * actual generic type arguments. + * + * @author Sebastien Deleuze + * @since 5.0 + */ +@FlowPreview +suspend inline fun ServerResponse.BodyBuilder.bodyAndAwait(flow: Flow): ServerResponse = + body(flow.asPublisher(), object : ParameterizedTypeReference() {}).awaitSingle() + /** * Coroutines variant of [ServerResponse.BodyBuilder.syncBody]. * @@ -83,6 +99,18 @@ suspend fun ServerResponse.HeadersBuilder>. suspend fun ServerResponse.BodyBuilder.bodyAndAwait(body: Any): ServerResponse = syncBody(body).awaitSingle() +/** + * Coroutines [Flow] based extension for [ServerResponse.BodyBuilder.body] providing a + * `bodyToServerSentEvents(Flow)` variant. This extension is not subject to type + * erasure and retains actual generic type arguments. + * + * @author Sebastien Deleuze + * @since 5.0 + */ +@FlowPreview +suspend inline fun ServerResponse.BodyBuilder.bodyToServerSentEventsAndAwait(flow: Flow): ServerResponse = + contentType(MediaType.TEXT_EVENT_STREAM).body(flow.asPublisher(), object : ParameterizedTypeReference() {}).awaitSingle() + /** * Coroutines variant of [ServerResponse.BodyBuilder.syncBody] without the sync prefix since it is ok to use it within diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensionsTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensionsTests.kt index b762fe7b61..dc5ee9f183 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensionsTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/ClientResponseExtensionsTests.kt @@ -19,6 +19,7 @@ package org.springframework.web.reactive.function.client import io.mockk.every import io.mockk.mockk import io.mockk.verify +import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals import org.junit.Assert.assertNull @@ -49,6 +50,13 @@ class ClientResponseExtensionsTests { verify { response.bodyToFlux(object : ParameterizedTypeReference>() {}) } } + @Test + @FlowPreview + fun `bodyToFlow with reified type parameters`() { + response.bodyToFlow>() + verify { response.bodyToFlux(object : ParameterizedTypeReference>() {}) } + } + @Test fun `toEntity with reified type parameters`() { response.toEntity>() diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt index b76339e142..e8e5b9413f 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt @@ -19,6 +19,8 @@ package org.springframework.web.reactive.function.client import io.mockk.every import io.mockk.mockk import io.mockk.verify +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals import org.junit.Test @@ -45,6 +47,14 @@ class WebClientExtensionsTests { verify { requestBodySpec.body(body, object : ParameterizedTypeReference>() {}) } } + @Test + @FlowPreview + fun `RequestBodySpec#body with Flow and reified type parameters`() { + val body = mockk>>() + requestBodySpec.body(body) + verify { requestBodySpec.body(ofType>>(), object : ParameterizedTypeReference>() {}) } + } + @Test fun `ResponseSpec#bodyToMono with reified type parameters`() { responseSpec.bodyToMono>() @@ -57,6 +67,13 @@ class WebClientExtensionsTests { verify { responseSpec.bodyToFlux(object : ParameterizedTypeReference>() {}) } } + @Test + @FlowPreview + fun `bodyToFlow with reified type parameters`() { + responseSpec.bodyToFlow>() + verify { responseSpec.bodyToFlux(object : ParameterizedTypeReference>() {}) } + } + @Test fun awaitExchange() { val response = mockk() diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensionsTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensionsTests.kt index c6eb34bb55..d840e5b69e 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensionsTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerRequestExtensionsTests.kt @@ -19,6 +19,7 @@ package org.springframework.web.reactive.function.client import io.mockk.every import io.mockk.mockk import io.mockk.verify +import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals import org.junit.Assert.assertNull @@ -52,6 +53,13 @@ class ServerRequestExtensionsTests { verify { request.bodyToFlux(object : ParameterizedTypeReference>() {}) } } + @Test + @FlowPreview + fun `bodyToFlow with reified type parameters`() { + request.bodyToFlow>() + verify { request.bodyToFlux(object : ParameterizedTypeReference>() {}) } + } + @Test fun awaitBody() { every { request.bodyToMono() } returns Mono.just("foo") diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensionsTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensionsTests.kt index 4356fd3bf8..99a5f4a9dc 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensionsTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/ServerResponseExtensionsTests.kt @@ -19,12 +19,16 @@ package org.springframework.web.reactive.function.server import io.mockk.every import io.mockk.mockk import io.mockk.verify +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals import org.junit.Test import org.reactivestreams.Publisher import org.springframework.core.ParameterizedTypeReference +import org.springframework.http.MediaType import org.springframework.http.MediaType.* +import reactor.core.publisher.Flux import reactor.core.publisher.Mono /** @@ -48,7 +52,7 @@ class ServerResponseExtensionsTests { fun `BodyBuilder#bodyToServerSentEvents with Publisher and reified type parameters`() { val body = mockk>>() bodyBuilder.bodyToServerSentEvents(body) - verify { bodyBuilder.contentType(TEXT_EVENT_STREAM) } + verify { bodyBuilder.contentType(TEXT_EVENT_STREAM).body(ofType>>(), object : ParameterizedTypeReference>() {}) } } @Test @@ -92,6 +96,32 @@ class ServerResponseExtensionsTests { } } + @Test + @FlowPreview + fun `BodyBuilder#body with Flow and reified type parameters`() { + val response = mockk() + val body = mockk>>() + every { bodyBuilder.body(ofType>>()) } returns Mono.just(response) + runBlocking { + bodyBuilder.bodyAndAwait(body) + } + verify { bodyBuilder.body(ofType>>(), object : ParameterizedTypeReference>() {}) } + } + + @Test + @FlowPreview + fun `BodyBuilder#bodyToServerSentEvents with Flow and reified type parameters`() { + val response = mockk() + val body = mockk>>() + every { bodyBuilder.contentType(ofType()) } returns bodyBuilder + every { bodyBuilder.body(ofType>>()) } returns Mono.just(response) + runBlocking { + bodyBuilder.bodyToServerSentEventsAndAwait(body) + } + verify { bodyBuilder.body(ofType>>(), object : ParameterizedTypeReference>() {}) } + verify { bodyBuilder.contentType(TEXT_EVENT_STREAM) } + } + @Test fun `renderAndAwait with a vararg parameter`() { val response = mockk()