Polishing

spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java

@@ -24,8 +24,6 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow;
 import java.util.function.Function;
 
-import kotlinx.coroutines.CompletableDeferredKt;
-import kotlinx.coroutines.Deferred;
 import org.reactivestreams.Publisher;
 import reactor.adapter.JdkFlowAdapter;
 import reactor.blockhound.BlockHound;
@@ -38,13 +36,14 @@ import org.springframework.util.ClassUtils;
 import org.springframework.util.ConcurrentReferenceHashMap;
 
 /**
- * A registry of adapters to adapt Reactive Streams {@link Publisher} to/from
- * various async/reactive types such as {@code CompletableFuture}, RxJava
- * {@code Flowable}, and others.
+ * A registry of adapters to adapt Reactive Streams {@link Publisher} to/from various
+ * async/reactive types such as {@code CompletableFuture}, RxJava {@code Flowable}, etc.
+ * This is designed to complement Spring's Reactor {@code Mono}/{@code Flux} support while
+ * also being usable without Reactor, e.g. just for {@code org.reactivestreams} bridging.
  *
- * <p>By default, depending on classpath availability, adapters are registered
- * for Reactor, RxJava 3, {@link CompletableFuture}, {@code Flow.Publisher},
- * and Kotlin Coroutines' {@code Deferred} and {@code Flow}.
+ * <p>By default, depending on classpath availability, adapters are registered for Reactor
+ * (including {@code CompletableFuture} and {@code Flow.Publisher} adapters), RxJava 3,
+ * Kotlin Coroutines' {@code Deferred} (bridged via Reactor) and SmallRye Mutiny 1.x.
  *
  * @author Rossen Stoyanchev
  * @author Sebastien Deleuze
@@ -304,9 +303,9 @@ public class ReactiveAdapterRegistry {
 		@SuppressWarnings("KotlinInternalInJava")
 		void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
-					ReactiveTypeDescriptor.singleOptionalValue(Deferred.class,
-							() -> CompletableDeferredKt.CompletableDeferred(null)),
-					source -> CoroutinesUtils.deferredToMono((Deferred<?>) source),
+					ReactiveTypeDescriptor.singleOptionalValue(kotlinx.coroutines.Deferred.class,
+							() -> kotlinx.coroutines.CompletableDeferredKt.CompletableDeferred(null)),
+					source -> CoroutinesUtils.deferredToMono((kotlinx.coroutines.Deferred<?>) source),
 					source -> CoroutinesUtils.monoToDeferred(Mono.from(source)));
 
 			registry.registerReactiveType(

spring-core/src/main/java/org/springframework/core/ReactiveTypeDescriptor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,7 +37,7 @@ public final class ReactiveTypeDescriptor {
 	private final boolean noValue;
 
 	@Nullable
-	private final Supplier<?> emptyValueSupplier;
+	private final Supplier<?> emptySupplier;
 
 	private final boolean deferred;
 
@@ -55,7 +55,7 @@ public final class ReactiveTypeDescriptor {
 		this.reactiveType = reactiveType;
 		this.multiValue = multiValue;
 		this.noValue = noValue;
-		this.emptyValueSupplier = emptySupplier;
+		this.emptySupplier = emptySupplier;
 		this.deferred = deferred;
 	}
 
@@ -89,7 +89,7 @@ public final class ReactiveTypeDescriptor {
 	 * Return {@code true} if the reactive type can complete with no values.
 	 */
 	public boolean supportsEmpty() {
-		return (this.emptyValueSupplier != null);
+		return (this.emptySupplier != null);
 	}
 
 	/**
@@ -97,8 +97,8 @@ public final class ReactiveTypeDescriptor {
 	 * <p>Use of this type implies {@link #supportsEmpty()} is {@code true}.
 	 */
 	public Object getEmptyValue() {
-		Assert.state(this.emptyValueSupplier != null, "Empty values not supported");
-		return this.emptyValueSupplier.get();
+		Assert.state(this.emptySupplier != null, "Empty values not supported");
+		return this.emptySupplier.get();
 	}
 
 	/**

spring-core/src/test/java/org/springframework/core/ReactiveAdapterRegistryTests.java

@@ -20,6 +20,7 @@ import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow;
 
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
@@ -27,6 +28,7 @@ import kotlinx.coroutines.Deferred;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
+import reactor.adapter.JdkFlowAdapter;
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -112,6 +114,16 @@ class ReactiveAdapterRegistryTests {
 			assertThat(((Mono<Integer>) target).block(ONE_SECOND)).isEqualTo(Integer.valueOf(1));
 		}
 
+		@Test
+		void toFlowPublisher() {
+			List<Integer> sequence = Arrays.asList(1, 2, 3);
+			Publisher<Integer> source = io.reactivex.rxjava3.core.Flowable.fromIterable(sequence);
+			Object target = getAdapter(Flow.Publisher.class).fromPublisher(source);
+			assertThat(target).isInstanceOf(Flow.Publisher.class);
+			assertThat(JdkFlowAdapter.flowPublisherToFlux((Flow.Publisher<Integer>) target)
+					.collectList().block(ONE_SECOND)).isEqualTo(sequence);
+		}
+
 		@Test
 		void toCompletableFuture() throws Exception {
 			Publisher<Integer> source = Flux.fromArray(new Integer[] {1, 2, 3});