From a8d6ba9965c0e084b7ed85fd11cb97a0af63bb2a Mon Sep 17 00:00:00 2001 From: Sebastien Deleuze Date: Thu, 4 Apr 2019 17:27:07 +0200 Subject: [PATCH] Add support for Coroutines Flow Flow is a Kotlin Coroutines related cold asynchronous stream of the data, that emits from zero to N (where N can be unbounded) values and completes normally or with an exception. It is conceptually the Coroutines equivalent of Flux with an extension oriented API design, easy custom operator capabilities and some suspending methods. This commit leverages Flow <-> Flux interoperability to support Flow on controller handler method parameters or return values, and also adds Flow based extensions to WebFlux.fn. It allows to reach a point when we can consider Spring Framework officially supports Coroutines even if some additional work remains to be done like adding interoperability between Reactor and Coroutines contexts. Flow is currently an experimental API that is expected to become final before Spring Framework 5.2 GA. Close gh-19975 --- build.gradle | 2 +- .../core/ReactiveAdapterRegistry.java | 17 +++++++++ .../KotlinReactiveAdapterRegistryTests.kt | 37 +++++++++++++++++++ .../client/ClientResponseExtensions.kt | 16 ++++++++ .../function/client/WebClientExtensions.kt | 32 +++++++++++++++- .../server/ServerRequestExtensions.kt | 16 ++++++++ .../server/ServerResponseExtensions.kt | 28 ++++++++++++++ .../client/ClientResponseExtensionsTests.kt | 8 ++++ .../client/WebClientExtensionsTests.kt | 17 +++++++++ .../server/ServerRequestExtensionsTests.kt | 8 ++++ .../server/ServerResponseExtensionsTests.kt | 32 +++++++++++++++- 11 files changed, 210 insertions(+), 3 deletions(-) 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()