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

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
+			);
+		}
 	}
 
 }

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<Int> = 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)!!
 	}