Add isDeferred to ReactiveTypeDescriptor

Closes gh-24995

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

@@ -218,7 +218,7 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 					source -> source);
 
 			registry.registerReactiveType(
-					ReactiveTypeDescriptor.singleOptionalValue(CompletionStage.class, EmptyCompletableFuture::new),
+					ReactiveTypeDescriptor.nonDeferredAsyncValue(CompletionStage.class, EmptyCompletableFuture::new),
 					source -> Mono.fromCompletionStage((CompletionStage<?>) source),
 					source -> Mono.from(source).toFuture()
 			);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -39,18 +39,24 @@ public final class ReactiveTypeDescriptor {
 	@Nullable
 	private final Supplier<?> emptyValueSupplier;
 
+	private final boolean deferred;
+
 
-	/**
-	 * Private constructor. See static factory methods.
-	 */
 	private ReactiveTypeDescriptor(Class<?> reactiveType, boolean multiValue, boolean noValue,
 			@Nullable Supplier<?> emptySupplier) {
 
+		this(reactiveType, multiValue, noValue, emptySupplier, true);
+	}
+
+	private ReactiveTypeDescriptor(Class<?> reactiveType, boolean multiValue, boolean noValue,
+			@Nullable Supplier<?> emptySupplier, boolean deferred) {
+
 		Assert.notNull(reactiveType, "'reactiveType' must not be null");
 		this.reactiveType = reactiveType;
 		this.multiValue = multiValue;
 		this.noValue = noValue;
 		this.emptyValueSupplier = emptySupplier;
+		this.deferred = deferred;
 	}
 
 
@@ -95,6 +101,16 @@ public Object getEmptyValue() {
 		return this.emptyValueSupplier.get();
 	}
 
+	/**
+	 * Whether the underlying operation is deferred and needs to be started
+	 * explicitly, e.g. via subscribing (or similar), or whether it is triggered
+	 * without the consumer having any control.
+	 * @since 5.1.16
+	 */
+	public boolean isDeferred() {
+		return this.deferred;
+	}
+
 
 	@Override
 	public boolean equals(@Nullable Object other) {
@@ -148,4 +164,15 @@ public static ReactiveTypeDescriptor noValue(Class<?> type, Supplier<?> emptySup
 		return new ReactiveTypeDescriptor(type, false, true, emptySupplier);
 	}
 
+	/**
+	 * The same as {@link #singleOptionalValue(Class, Supplier)} but for a
+	 * non-deferred, async type such as {@link java.util.concurrent.CompletableFuture}.
+	 * @param type the reactive type
+	 * @param emptySupplier a supplier of an empty-value instance of the reactive type
+	 * @since 5.1.16
+	 */
+	public static ReactiveTypeDescriptor nonDeferredAsyncValue(Class<?> type, Supplier<?> emptySupplier) {
+		return new ReactiveTypeDescriptor(type, false, false, emptySupplier, false);
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -265,9 +265,26 @@ void completableFutureToPublisher() {
 		assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
 	}
 
+	@Test
+	void deferred() {
+		assertThat(getAdapter(CompletableFuture.class).getDescriptor().isDeferred()).isEqualTo(false);
+
+		assertThat(getAdapter(Mono.class).getDescriptor().isDeferred()).isEqualTo(true);
+		assertThat(getAdapter(Flux.class).getDescriptor().isDeferred()).isEqualTo(true);
+
+		assertThat(getAdapter(io.reactivex.Completable.class).getDescriptor().isDeferred()).isEqualTo(true);
+		assertThat(getAdapter(io.reactivex.Single.class).getDescriptor().isDeferred()).isEqualTo(true);
+		assertThat(getAdapter(io.reactivex.Flowable.class).getDescriptor().isDeferred()).isEqualTo(true);
+		assertThat(getAdapter(io.reactivex.Observable.class).getDescriptor().isDeferred()).isEqualTo(true);
+
+		assertThat(getAdapter(Deferred.class).getDescriptor().isDeferred()).isEqualTo(true);
+		assertThat(getAdapter(kotlinx.coroutines.flow.Flow.class).getDescriptor().isDeferred()).isEqualTo(true);
+	}
 
 	private ReactiveAdapter getAdapter(Class<?> reactiveType) {
-		return this.registry.getAdapter(reactiveType);
+		ReactiveAdapter adapter = this.registry.getAdapter(reactiveType);
+		assertThat(adapter).isNotNull();
+		return adapter;
 	}
 
 }