Deprecate RxJava 2 in favor of RxJava 3

Closes gh-27474

Author: jhoeller

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

@@ -44,12 +44,11 @@
  * {@code Observable}, and others.
  *
  * <p>By default, depending on classpath availability, adapters are registered
- * for Reactor, RxJava 2/3, or RxJava 1 (+ RxJava Reactive Streams bridge),
- * {@link CompletableFuture}, Java 9+ {@code Flow.Publisher}, and Kotlin
- * Coroutines' {@code Deferred} and {@code Flow}.
+ * for Reactor, RxJava 3, {@link CompletableFuture}, {@code Flow.Publisher},
+ * and Kotlin Coroutines' {@code Deferred} and {@code Flow}.
  *
- * <p><strong>Note:</strong> As of Spring Framework 5.3, support for RxJava 1.x
- * is deprecated in favor of RxJava 2 and 3.
+ * <p><strong>Note:</strong> As of Spring Framework 5.3.11, support for
+ * RxJava 1.x and 2.x is deprecated in favor of RxJava 3.
  *
  * @author Rossen Stoyanchev
  * @author Sebastien Deleuze
@@ -103,15 +102,13 @@ public ReactiveAdapterRegistry() {
 			}
 		}
 
-		// RxJava1 (deprecated)
+		// RxJava
 		if (rxjava1Present) {
 			new RxJava1Registrar().registerAdapters(this);
 		}
-		// RxJava2
 		if (rxjava2Present) {
 			new RxJava2Registrar().registerAdapters(this);
 		}
-		// RxJava3
 		if (rxjava3Present) {
 			new RxJava3Registrar().registerAdapters(this);
 		}
@@ -219,6 +216,29 @@ public static ReactiveAdapterRegistry getSharedInstance() {
 	}
 
 
+	/**
+	 * ReactiveAdapter variant that wraps adapted Publishers as {@link Flux} or
+	 * {@link Mono} depending on {@link ReactiveTypeDescriptor#isMultiValue()}.
+	 * This is important in places where only the stream and stream element type
+	 * information is available like encoders and decoders.
+	 */
+	private static class ReactorAdapter extends ReactiveAdapter {
+
+		ReactorAdapter(ReactiveTypeDescriptor descriptor,
+				Function<Object, Publisher<?>> toPublisherFunction,
+				Function<Publisher<?>, Object> fromPublisherFunction) {
+
+			super(descriptor, toPublisherFunction, fromPublisherFunction);
+		}
+
+		@Override
+		public <T> Publisher<T> toPublisher(@Nullable Object source) {
+			Publisher<T> publisher = super.toPublisher(source);
+			return (isMultiValue() ? Flux.from(publisher) : Mono.from(publisher));
+		}
+	}
+
+
 	private static class ReactorRegistrar {
 
 		void registerAdapters(ReactiveAdapterRegistry registry) {
@@ -227,8 +247,7 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
 					source -> (Mono<?>) source,
-					Mono::from
-			);
+					Mono::from);
 
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
@@ -243,8 +262,15 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.nonDeferredAsyncValue(CompletionStage.class, EmptyCompletableFuture::new),
 					source -> Mono.fromCompletionStage((CompletionStage<?>) source),
-					source -> Mono.from(source).toFuture()
-			);
+					source -> Mono.from(source).toFuture());
+		}
+	}
+
+
+	private static class EmptyCompletableFuture<T> extends CompletableFuture<T> {
+
+		EmptyCompletableFuture() {
+			complete(null);
 		}
 	}
 
@@ -268,8 +294,7 @@ void registerAdapter(ReactiveAdapterRegistry registry) {
 				registry.registerReactiveType(
 						ReactiveTypeDescriptor.multiValue(publisherClass, () -> emptyFlow),
 						source -> (Publisher<?>) ReflectionUtils.invokeMethod(toFluxMethod, null, source),
-						publisher -> ReflectionUtils.invokeMethod(toFlowMethod, null, publisher)
-				);
+						publisher -> ReflectionUtils.invokeMethod(toFlowMethod, null, publisher));
 			}
 			catch (Throwable ex) {
 				// Ignore
@@ -284,18 +309,17 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(rx.Observable.class, rx.Observable::empty),
 					source -> RxReactiveStreams.toPublisher((rx.Observable<?>) source),
-					RxReactiveStreams::toObservable
-			);
+					RxReactiveStreams::toObservable);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.singleRequiredValue(rx.Single.class),
 					source -> RxReactiveStreams.toPublisher((rx.Single<?>) source),
-					RxReactiveStreams::toSingle
-			);
+					RxReactiveStreams::toSingle);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.noValue(rx.Completable.class, rx.Completable::complete),
 					source -> RxReactiveStreams.toPublisher((rx.Completable) source),
-					RxReactiveStreams::toCompletable
-			);
+					RxReactiveStreams::toCompletable);
 		}
 	}
 
@@ -306,29 +330,28 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
 					source -> (io.reactivex.Flowable<?>) source,
-					io.reactivex.Flowable::fromPublisher
-			);
+					io.reactivex.Flowable::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
 					source -> ((io.reactivex.Observable<?>) source).toFlowable(io.reactivex.BackpressureStrategy.BUFFER),
-					io.reactivex.Observable::fromPublisher
-			);
+					io.reactivex.Observable::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
 					source -> ((io.reactivex.Single<?>) source).toFlowable(),
-					io.reactivex.Single::fromPublisher
-			);
+					io.reactivex.Single::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
 					source -> ((io.reactivex.Maybe<?>) source).toFlowable(),
 					source -> io.reactivex.Flowable.fromPublisher(source)
-							.toObservable().singleElement()
-			);
+							.toObservable().singleElement());
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
 					source -> ((io.reactivex.Completable) source).toFlowable(),
-					io.reactivex.Completable::fromPublisher
-			);
+					io.reactivex.Completable::fromPublisher);
 		}
 	}
 
@@ -341,66 +364,34 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 							io.reactivex.rxjava3.core.Flowable.class,
 							io.reactivex.rxjava3.core.Flowable::empty),
 					source -> (io.reactivex.rxjava3.core.Flowable<?>) source,
-					io.reactivex.rxjava3.core.Flowable::fromPublisher
-			);
+					io.reactivex.rxjava3.core.Flowable::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(
 							io.reactivex.rxjava3.core.Observable.class,
 							io.reactivex.rxjava3.core.Observable::empty),
 					source -> ((io.reactivex.rxjava3.core.Observable<?>) source).toFlowable(
 							io.reactivex.rxjava3.core.BackpressureStrategy.BUFFER),
-					io.reactivex.rxjava3.core.Observable::fromPublisher
-			);
+					io.reactivex.rxjava3.core.Observable::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.rxjava3.core.Single.class),
 					source -> ((io.reactivex.rxjava3.core.Single<?>) source).toFlowable(),
-					io.reactivex.rxjava3.core.Single::fromPublisher
-			);
+					io.reactivex.rxjava3.core.Single::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.singleOptionalValue(
 							io.reactivex.rxjava3.core.Maybe.class,
 							io.reactivex.rxjava3.core.Maybe::empty),
 					source -> ((io.reactivex.rxjava3.core.Maybe<?>) source).toFlowable(),
-					io.reactivex.rxjava3.core.Maybe::fromPublisher
-			);
+					io.reactivex.rxjava3.core.Maybe::fromPublisher);
+
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.noValue(
 							io.reactivex.rxjava3.core.Completable.class,
 							io.reactivex.rxjava3.core.Completable::complete),
 					source -> ((io.reactivex.rxjava3.core.Completable) source).toFlowable(),
-					io.reactivex.rxjava3.core.Completable::fromPublisher
-			);
-		}
-	}
-
-
-	/**
-	 * ReactiveAdapter variant that wraps adapted Publishers as {@link Flux} or
-	 * {@link Mono} depending on {@link ReactiveTypeDescriptor#isMultiValue()}.
-	 * This is important in places where only the stream and stream element type
-	 * information is available like encoders and decoders.
-	 */
-	private static class ReactorAdapter extends ReactiveAdapter {
-
-		ReactorAdapter(ReactiveTypeDescriptor descriptor,
-				Function<Object, Publisher<?>> toPublisherFunction,
-				Function<Publisher<?>, Object> fromPublisherFunction) {
-
-			super(descriptor, toPublisherFunction, fromPublisherFunction);
-		}
-
-		@Override
-		public <T> Publisher<T> toPublisher(@Nullable Object source) {
-			Publisher<T> publisher = super.toPublisher(source);
-			return (isMultiValue() ? Flux.from(publisher) : Mono.from(publisher));
-		}
-	}
-
-
-	private static class EmptyCompletableFuture<T> extends CompletableFuture<T> {
-
-		EmptyCompletableFuture() {
-			complete(null);
+					io.reactivex.rxjava3.core.Completable::fromPublisher);
 		}
 	}
 
@@ -418,8 +409,7 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(kotlinx.coroutines.flow.Flow.class, kotlinx.coroutines.flow.FlowKt::emptyFlow),
 					source -> kotlinx.coroutines.reactor.ReactorFlowKt.asFlux((kotlinx.coroutines.flow.Flow<?>) source),
-					kotlinx.coroutines.reactive.ReactiveFlowKt::asFlow
-			);
+					kotlinx.coroutines.reactive.ReactiveFlowKt::asFlow);
 		}
 	}
 
@@ -432,16 +422,14 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 							io.smallrye.mutiny.Uni.class,
 							() -> io.smallrye.mutiny.Uni.createFrom().nothing()),
 					uni -> ((io.smallrye.mutiny.Uni<?>) uni).convert().toPublisher(),
-					publisher -> io.smallrye.mutiny.Uni.createFrom().publisher(publisher)
-			);
+					publisher -> io.smallrye.mutiny.Uni.createFrom().publisher(publisher));
 
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(
 							io.smallrye.mutiny.Multi.class,
 							() -> io.smallrye.mutiny.Multi.createFrom().empty()),
 					multi -> (io.smallrye.mutiny.Multi<?>) multi,
-					publisher -> io.smallrye.mutiny.Multi.createFrom().publisher(publisher)
-			);
+					publisher -> io.smallrye.mutiny.Multi.createFrom().publisher(publisher));
 		}
 	}
 
@@ -459,7 +447,6 @@ public static class SpringCoreBlockHoundIntegration implements BlockHoundIntegra
 
 		@Override
 		public void applyTo(BlockHound.Builder builder) {
-
 			// Avoid hard references potentially anywhere in spring-core (no need for structural dependency)
 
 			builder.allowBlockingCallsInside(