Merge pull request #1248 from rstoyanchev/14902-reactive-adapters

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

@@ -16,99 +16,81 @@
 
 package org.springframework.core;
 
+import java.util.Optional;
+import java.util.function.Function;
+
 import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+
+import org.springframework.util.Assert;
 
 /**
- * Contract for adapting to and from {@link Flux} and {@link Mono}.
+ * Adapt a Reactive Streams {@link Publisher} to and from an async/reactive type
+ * such as {@code CompletableFuture}, an RxJava {@code Observable}, etc.
  *
- * <p>An adapter supports a specific adaptee type whose stream semantics
- * can be checked via {@link #getDescriptor()}.
- *
- * <p>Use the {@link ReactiveAdapterRegistry} to obtain an adapter for a
- * supported adaptee type or to register additional adapters.
+ * <p>Use the {@link ReactiveAdapterRegistry} to register reactive types and
+ * obtain adapters from.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
  */
-public interface ReactiveAdapter {
+public class ReactiveAdapter {
 
-	/**
-	 * Return a descriptor with further information about the adaptee.
-	 */
-	Descriptor getDescriptor();
+	private final ReactiveTypeDescriptor descriptor;
 
-	/**
-	 * Adapt the given Object to a {@link Mono}
-	 * @param source the source object to adapt
-	 * @return the resulting {@link Mono} possibly empty
-	 */
-	<T> Mono<T> toMono(Object source);
+	private final Function<Object, Publisher<?>> toPublisherFunction;
 
-	/**
-	 * Adapt the given Object to a {@link Flux}.
-	 * @param source the source object to adapt
-	 * @return the resulting {@link Flux} possibly empty
-	 */
-	<T> Flux<T> toFlux(Object source);
-
-	/**
-	 * Adapt the given Object to a Publisher.
-	 * @param source the source object to adapt
-	 * @return the resulting {@link Mono} or {@link Flux} possibly empty
-	 */
-	<T> Publisher<T> toPublisher(Object source);
-
-	/**
-	 * Adapt the given Publisher to the target adaptee.
-	 * @param publisher the publisher to adapt
-	 * @return the resulting adaptee
-	 */
-	Object fromPublisher(Publisher<?> publisher);
+	private final Function<Publisher<?>, Object> fromPublisherFunction;
 
 
 	/**
-	 * A descriptor with information about the adaptee stream semantics.
+	 * Constructor for an adapter with functions to convert the target reactive
+	 * or async type to and from a Reactive Streams Publisher.
+	 * @param descriptor the reactive type descriptor
+	 * @param toPublisherFunction adapter to a Publisher
+	 * @param fromPublisherFunction adapter from a Publisher
 	 */
-	class Descriptor {
+	public ReactiveAdapter(ReactiveTypeDescriptor descriptor,
+			Function<Object, Publisher<?>> toPublisherFunction,
+			Function<Publisher<?>, Object> fromPublisherFunction) {
 
-		private final boolean isMultiValue;
+		Assert.notNull(descriptor, "'descriptor' is required");
+		Assert.notNull(toPublisherFunction, "'toPublisherFunction' is required");
+		Assert.notNull(fromPublisherFunction, "'fromPublisherFunction' is required");
 
-		private final boolean supportsEmpty;
+		this.descriptor = descriptor;
+		this.toPublisherFunction = toPublisherFunction;
+		this.fromPublisherFunction = fromPublisherFunction;
+	}
 
-		private final boolean isNoValue;
 
-		public Descriptor(boolean isMultiValue, boolean canBeEmpty, boolean isNoValue) {
-			this.isMultiValue = isMultiValue;
-			this.supportsEmpty = canBeEmpty;
-			this.isNoValue = isNoValue;
+	/**
+	 * Return the descriptor of the reactive type for the adapter.
+	 */
+	public ReactiveTypeDescriptor getDescriptor() {
+		return this.descriptor;
+	}
+
+	/**
+	 * Adapt the given instance to a Reactive Streams Publisher.
+	 * @param source the source object to adapt from
+	 * @return the Publisher repesenting the adaptation
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> Publisher<T> toPublisher(Object source) {
+		source = (source instanceof Optional ? ((Optional<?>) source).orElse(null) : source);
+		if (source == null) {
+			source = getDescriptor().getEmptyValue();
 		}
+		return (Publisher<T>) this.toPublisherFunction.apply(source);
+	}
 
-		/**
-		 * Return {@code true} if the adaptee implies 0..N values can be produced
-		 * and is therefore a good fit to adapt to {@link Flux}. A {@code false}
-		 * return value implies the adaptee will produce 1 value at most and is
-		 * therefore a good fit for {@link Mono}.
-		 */
-		public boolean isMultiValue() {
-			return this.isMultiValue;
-		}
-
-		/**
-		 * Return {@code true} if the adaptee can complete without values.
-		 */
-		public boolean supportsEmpty() {
-			return this.supportsEmpty;
-		}
-
-		/**
-		 * Return {@code true} if the adaptee implies no values will be produced,
-		 * i.e. providing only completion or error signal.
-		 */
-		public boolean isNoValue() {
-			return this.isNoValue;
-		}
+	/**
+	 * Adapt from the given Reactive Streams Publisher.
+	 * @param publisher the publisher to adapt from
+	 * @return the reactive type instance representing the adapted publisher
+	 */
+	public Object fromPublisher(Publisher<?> publisher) {
+		return (publisher != null ? this.fromPublisherFunction.apply(publisher) : null);
 	}
 
 }

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

@@ -16,32 +16,32 @@
 
 package org.springframework.core;
 
-import java.util.LinkedHashMap;
-import java.util.Map;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
 import io.reactivex.BackpressureStrategy;
+import io.reactivex.Completable;
 import io.reactivex.Flowable;
 import io.reactivex.Maybe;
+import io.reactivex.Observable;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import rx.Completable;
-import rx.Observable;
 import rx.RxReactiveStreams;
-import rx.Single;
 
 import org.springframework.util.ClassUtils;
 
 /**
- * A registry of adapters to adapt to {@link Flux} and {@link Mono}.
+ * A registry of adapters to adapt a Reactive Streams {@link Publisher} to/from
+ * various async/reactive types such as {@code CompletableFuture}, RxJava
+ * {@code Observable}, and others.
  *
- * <p>By default there are adapters for {@link CompletableFuture}, RxJava 1, and
- * also for a any Reactive Streams {@link Publisher}. Additional adapters can be
- * registered via {@link #registerFluxAdapter} and {@link #registerMonoAdapter}.
+ * <p>By default, depending on classpath availability, adapters are registered
+ * for Reactor, RxJava 1, RxJava 2 types, and {@link CompletableFuture}.
  *
  * @author Rossen Stoyanchev
  * @author Sebastien Deleuze
@@ -49,6 +49,9 @@ import org.springframework.util.ClassUtils;
  */
 public class ReactiveAdapterRegistry {
 
+	private static final boolean reactorPresent =
+			ClassUtils.isPresent("reactor.core.publisher.Flux", ReactiveAdapterRegistry.class.getClassLoader());
+
 	private static final boolean rxJava1Present =
 			ClassUtils.isPresent("rx.Observable", ReactiveAdapterRegistry.class.getClassLoader());
 
@@ -58,257 +61,180 @@ public class ReactiveAdapterRegistry {
 	private static final boolean rxJava2Present =
 			ClassUtils.isPresent("io.reactivex.Flowable", ReactiveAdapterRegistry.class.getClassLoader());
 
-	private final Map<Class<?>, ReactiveAdapter> adapterMap = new LinkedHashMap<>(4);
+
+	private final List<ReactiveAdapter> adapters = new ArrayList<>(32);
 
 
 	/**
 	 * Create a registry and auto-register default adapters.
 	 */
 	public ReactiveAdapterRegistry() {
-		// Flux and Mono ahead of Publisher...
-		registerMonoAdapter(Mono.class,
-				source -> (Mono<?>) source, source -> source,
-				new ReactiveAdapter.Descriptor(false, true, false));
-		registerFluxAdapter(
-				Flux.class, source -> (Flux<?>) source, source -> source);
-		registerFluxAdapter(
-				Publisher.class, source -> Flux.from((Publisher<?>) source), source -> source);
 
-		registerMonoAdapter(CompletableFuture.class,
-				source -> Mono.fromFuture((CompletableFuture<?>) source), Mono::toFuture,
-				new ReactiveAdapter.Descriptor(false, true, false)
-		);
+		if (reactorPresent) {
+			new ReactorRegistrar().registerAdapters(this);
+		}
 
 		if (rxJava1Present && rxJava1Adapter) {
-			new RxJava1AdapterRegistrar().register(this);
+			new RxJava1Registrar().registerAdapters(this);
 		}
+
 		if (rxJava2Present) {
-			new RxJava2AdapterRegistrar().register(this);
+			new RxJava2Registrar().registerAdapters(this);
 		}
 	}
 
 
 	/**
-	 * Register an adapter for adapting to and from a {@link Mono}.
-	 * <p>The provided functions can assume that input will never be {@code null}
-	 * and also that any {@link Optional} wrapper is unwrapped.
+	 * Register a reactive type along with functions to adapt to and from a
+	 * Reactive Streams {@link Publisher}. The functions can assume their
+	 * input is never be {@code null} nor {@link Optional}.
 	 */
-	public void registerMonoAdapter(Class<?> adapteeType, Function<Object, Mono<?>> toAdapter,
-			Function<Mono<?>, Object> fromAdapter, ReactiveAdapter.Descriptor descriptor) {
+	public void registerReactiveType(ReactiveTypeDescriptor descriptor,
+			Function<Object, Publisher<?>> toAdapter, Function<Publisher<?>, Object> fromAdapter) {
 
-		this.adapterMap.put(adapteeType, new MonoReactiveAdapter(toAdapter, fromAdapter, descriptor));
+		if (reactorPresent) {
+			this.adapters.add(new ReactorAdapter(descriptor, toAdapter, fromAdapter));
+		}
+		else {
+			this.adapters.add(new ReactiveAdapter(descriptor, toAdapter, fromAdapter));
+		}
 	}
 
 	/**
-	 * Register an adapter for adapting to and from a {@link Flux}.
-	 * <p>The provided functions can assume that input will never be {@code null}
-	 * and also that any {@link Optional} wrapper is unwrapped.
+	 * Get the adapter to use to adapt from the given reactive type.
 	 */
-	public void registerFluxAdapter(Class<?> adapteeType, Function<Object, Flux<?>> toAdapter,
-			Function<Flux<?>, Object> fromAdapter) {
-
-		this.adapterMap.put(adapteeType, new FluxReactiveAdapter(toAdapter, fromAdapter));
-	}
-
-
-	/**
-	 * Get the adapter for the given adaptee type to adapt from.
-	 */
-	public ReactiveAdapter getAdapterFrom(Class<?> adapteeType) {
-		return getAdapterFrom(adapteeType, null);
+	public ReactiveAdapter getAdapterFrom(Class<?> reactiveType) {
+		return getAdapterFrom(reactiveType, null);
 	}
 
 	/**
-	 * Get the adapter for the given adaptee type to adapt from.
-	 * If the instance is not {@code null} its actual type is used to check.
+	 * Get the adapter to use to adapt from the given reactive type. Or if the
+	 * "source" object is not {@code null} its actual type is used instead.
 	 */
-	public ReactiveAdapter getAdapterFrom(Class<?> adapteeType, Object adaptee) {
-		Class<?> actualType = getActualType(adapteeType, adaptee);
-		return getAdapterInternal(supportedType -> supportedType.isAssignableFrom(actualType));
+	public ReactiveAdapter getAdapterFrom(Class<?> reactiveType, Object source) {
+		source = (source instanceof Optional ? ((Optional<?>) source).orElse(null) : source);
+		Class<?> clazz = (source != null ? source.getClass() : reactiveType);
+		return getAdapter(type -> type.isAssignableFrom(clazz));
 	}
 
 	/**
-	 * Get the adapter for the given adaptee type to adapt to.
+	 * Get the adapter for the given reactive type to adapt to.
 	 */
-	public ReactiveAdapter getAdapterTo(Class<?> adapteeType) {
-		return getAdapterTo(adapteeType, null);
+	public ReactiveAdapter getAdapterTo(Class<?> reactiveType) {
+		return getAdapter(reactiveType::equals);
 	}
 
-	/**
-	 * Get the adapter for the given adaptee type to adapt to.
-	 * If the instance is not {@code null} its actual type is used to check.
-	 */
-	public ReactiveAdapter getAdapterTo(Class<?> adapteeType, Object adaptee) {
-		Class<?> actualType = getActualType(adapteeType, adaptee);
-		return getAdapterInternal(supportedType -> supportedType.equals(actualType));
-	}
-
-	private ReactiveAdapter getAdapterInternal(Predicate<Class<?>> adapteeTypePredicate) {
-		return this.adapterMap.keySet().stream()
-				.filter(adapteeTypePredicate)
-				.map(this.adapterMap::get)
+	private ReactiveAdapter getAdapter(Predicate<Class<?>> predicate) {
+		return this.adapters.stream()
+				.filter(adapter -> predicate.test(adapter.getDescriptor().getReactiveType()))
 				.findFirst()
 				.orElse(null);
 	}
 
 
-	private static Class<?> getActualType(Class<?> adapteeType, Object adaptee) {
-		adaptee = unwrapOptional(adaptee);
-		return (adaptee != null ? adaptee.getClass() : adapteeType);
-	}
+	private static class ReactorRegistrar {
 
-	private static Object unwrapOptional(Object value) {
-		return (value instanceof Optional ? ((Optional<?>) value).orElse(null) : value);
-	}
+		void registerAdapters(ReactiveAdapterRegistry registry) {
 
+			// Flux and Mono ahead of Publisher...
 
-	@SuppressWarnings("unchecked")
-	private static class MonoReactiveAdapter implements ReactiveAdapter {
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
+					source -> (Mono<?>) source,
+					Mono::from
+			);
 
-		private final Function<Object, Mono<?>> toAdapter;
+			registry.registerReactiveType(ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
+					source -> (Flux<?>) source,
+					Flux::from);
 
-		private final Function<Mono<?>, Object> fromAdapter;
+			registry.registerReactiveType(ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty),
+					source -> (Publisher<?>) source,
+					source -> source);
 
-		private final Descriptor descriptor;
-
-
-		MonoReactiveAdapter(Function<Object, Mono<?>> to, Function<Mono<?>, Object> from, Descriptor descriptor) {
-			this.toAdapter = to;
-			this.fromAdapter = from;
-			this.descriptor = descriptor;
-		}
-
-		@Override
-		public Descriptor getDescriptor() {
-			return this.descriptor;
-		}
-
-		@Override
-		public <T> Mono<T> toMono(Object source) {
-			source = unwrapOptional(source);
-			if (source == null) {
-				return Mono.empty();
-			}
-			return (Mono<T>) this.toAdapter.apply(source);
-		}
-
-		@Override
-		public <T> Flux<T> toFlux(Object source) {
-			source = unwrapOptional(source);
-			if (source == null) {
-				return Flux.empty();
-			}
-			return (Flux<T>) this.toMono(source).flux();
-		}
-
-		@Override
-		public <T> Publisher<T> toPublisher(Object source) {
-			return toMono(source);
-		}
-
-		@Override
-		public Object fromPublisher(Publisher<?> source) {
-			return (source != null ? this.fromAdapter.apply((Mono<?>) source) : null);
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.singleOptionalValue(CompletableFuture.class, () -> {
+						CompletableFuture<?> empty = new CompletableFuture<>();
+						empty.complete(null);
+						return empty;
+					}),
+					source -> Mono.fromFuture((CompletableFuture<?>) source),
+					source -> Mono.from(source).toFuture()
+			);
 		}
 	}
 
-	@SuppressWarnings("unchecked")
-	private static class FluxReactiveAdapter implements ReactiveAdapter {
+	private static class RxJava1Registrar {
 
-		private final Function<Object, Flux<?>> toAdapter;
-
-		private final Function<Flux<?>, Object> fromAdapter;
-
-		private final Descriptor descriptor = new Descriptor(true, true, false);
-
-
-		FluxReactiveAdapter(Function<Object, Flux<?>> to, Function<Flux<?>, Object> from) {
-			this.toAdapter = to;
-			this.fromAdapter = from;
-		}
-
-		@Override
-		public Descriptor getDescriptor() {
-			return this.descriptor;
-		}
-
-		@Override
-		public <T> Mono<T> toMono(Object source) {
-			source = unwrapOptional(source);
-			if (source == null) {
-				return Mono.empty();
-			}
-			return (Mono<T>) this.toAdapter.apply(source).next();
-		}
-
-		@Override
-		public <T> Flux<T> toFlux(Object source) {
-			source = unwrapOptional(source);
-			if (source == null) {
-				return Flux.empty();
-			}
-			return (Flux<T>) this.toAdapter.apply(source);
-		}
-
-		@Override
-		public <T> Publisher<T> toPublisher(Object source) {
-			return toFlux(source);
-		}
-
-		@Override
-		public Object fromPublisher(Publisher<?> source) {
-			return (source != null ? this.fromAdapter.apply((Flux<?>) source) : null);
-		}
-	}
-
-
-	private static class RxJava1AdapterRegistrar {
-
-		public void register(ReactiveAdapterRegistry registry) {
-			registry.registerFluxAdapter(Observable.class,
-					source -> Flux.from(RxReactiveStreams.toPublisher((Observable<?>) source)),
+		void registerAdapters(ReactiveAdapterRegistry registry) {
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.multiValue(rx.Observable.class, rx.Observable::empty),
+					source -> RxReactiveStreams.toPublisher((rx.Observable<?>) source),
 					RxReactiveStreams::toObservable
 			);
-			registry.registerMonoAdapter(Single.class,
-					source -> Mono.from(RxReactiveStreams.toPublisher((Single<?>) source)),
-					RxReactiveStreams::toSingle,
-					new ReactiveAdapter.Descriptor(false, false, false)
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.singleRequiredValue(rx.Single.class),
+					source -> RxReactiveStreams.toPublisher((rx.Single<?>) source),
+					RxReactiveStreams::toSingle
 			);
-			registry.registerMonoAdapter(Completable.class,
-					source -> Mono.from(RxReactiveStreams.toPublisher((Completable) source)),
-					RxReactiveStreams::toCompletable,
-					new ReactiveAdapter.Descriptor(false, true, true)
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.noValue(rx.Completable.class, Completable::complete),
+					source -> RxReactiveStreams.toPublisher((rx.Completable) source),
+					RxReactiveStreams::toCompletable
 			);
 		}
 	}
 
-	private static class RxJava2AdapterRegistrar {
+	private static class RxJava2Registrar {
 
-		public void register(ReactiveAdapterRegistry registry) {
-			registry.registerFluxAdapter(Flowable.class,
-					source -> Flux.from((Flowable<?>) source),
+		void registerAdapters(ReactiveAdapterRegistry registry) {
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.multiValue(Flowable.class, Flowable::empty),
+					source -> (Flowable<?>) source,
 					source-> Flowable.fromPublisher(source)
 			);
-			registry.registerFluxAdapter(io.reactivex.Observable.class,
-					source -> Flux.from(((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER)),
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.multiValue(Observable.class, Observable::empty),
+					source -> ((Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
 					source -> Flowable.fromPublisher(source).toObservable()
 			);
-			registry.registerMonoAdapter(io.reactivex.Single.class,
-					source -> Mono.from(((io.reactivex.Single<?>) source).toFlowable()),
-					source -> Flowable.fromPublisher(source).toObservable().singleElement().toSingle(),
-					new ReactiveAdapter.Descriptor(false, false, false)
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
+					source -> ((io.reactivex.Single<?>) source).toFlowable(),
+					source -> Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
 			);
-			registry.registerMonoAdapter(Maybe.class,
-					source -> Mono.from(((Maybe<?>) source).toFlowable()),
-					source -> Flowable.fromPublisher(source).toObservable().singleElement(),
-					new ReactiveAdapter.Descriptor(false, true, false)
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.singleOptionalValue(Maybe.class, Maybe::empty),
+					source -> ((Maybe<?>) source).toFlowable(),
+					source -> Flowable.fromPublisher(source).toObservable().singleElement()
 			);
-			registry.registerMonoAdapter(io.reactivex.Completable.class,
-					source -> Mono.from(((io.reactivex.Completable) source).toFlowable()),
-					source -> Flowable.fromPublisher(source).toObservable().ignoreElements(),
-					new ReactiveAdapter.Descriptor(false, true, true)
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.noValue(Completable.class, Completable::complete),
+					source -> ((Completable) source).toFlowable(),
+					source -> Flowable.fromPublisher(source).toObservable().ignoreElements()
 			);
 		}
 	}
 
+	/**
+	 * Extension of ReactiveAdapter that wraps adapted (raw) Publisher's as
+	 * {@link Flux} or {@link Mono} depending on the underlying reactive type's
+	 * stream semantics.
+	 */
+	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(Object source) {
+			Publisher<T> publisher = super.toPublisher(source);
+			return (getDescriptor().isMultiValue() ? Flux.from(publisher) : Mono.from(publisher));
+		}
+	}
+
 }

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

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.core;
+
+import java.util.function.Supplier;
+
+import org.springframework.util.Assert;
+
+/**
+ * Descriptor for a reactive type with information its stream semantics, i.e.
+ * how many values it can produce.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.0
+ */
+public class ReactiveTypeDescriptor {
+
+	private final Class<?> reactiveType;
+
+	private final Supplier<?> emptyValueSupplier;
+
+	private final boolean multiValue;
+
+	private final boolean supportsEmpty;
+
+	private final boolean noValue;
+
+
+	/**
+	 * Private constructor. See static factory methods.
+	 */
+	private ReactiveTypeDescriptor(Class<?> reactiveType, Supplier<?> emptySupplier,
+			boolean multiValue, boolean canBeEmpty, boolean noValue) {
+
+		Assert.notNull(reactiveType, "'reactiveType' must not be null");
+		Assert.isTrue(!canBeEmpty || emptySupplier != null, "Empty value supplier is required.");
+		this.reactiveType = reactiveType;
+		this.emptyValueSupplier = emptySupplier;
+		this.multiValue = multiValue;
+		this.supportsEmpty = canBeEmpty;
+		this.noValue = noValue;
+	}
+
+
+	/**
+	 * Return the reactive type the descriptor was created for.
+	 */
+	public Class<?> getReactiveType() {
+		return this.reactiveType;
+	}
+
+	/**
+	 * Return an empty-value instance for the underlying reactive or async type.
+	 * Use of this type implies {@link #supportsEmpty()} is true.
+	 */
+	public Object getEmptyValue() {
+		Assert.isTrue(supportsEmpty(), "Empty values not supported.");
+		return this.emptyValueSupplier.get();
+	}
+
+	/**
+	 * Return {@code true} if the reactive type can produce more than 1 value
+	 * can be produced and is therefore a good fit to adapt to {@code Flux}.
+	 * A {@code false} return value implies the reactive type can produce 1
+	 * value at most and is therefore a good fit to adapt to {@code Mono}.
+	 */
+	public boolean isMultiValue() {
+		return this.multiValue;
+	}
+
+	/**
+	 * Return {@code true} if the reactive type can complete with no values.
+	 */
+	public boolean supportsEmpty() {
+		return this.supportsEmpty;
+	}
+
+	/**
+	 * Return {@code true} if the reactive type does not produce any values and
+	 * only provides completion and error signals.
+	 */
+	public boolean isNoValue() {
+		return this.noValue;
+	}
+
+
+	@Override
+	public boolean equals(Object other) {
+		if (this == other) {
+			return true;
+		}
+		if (other == null || getClass() != other.getClass()) {
+			return false;
+		}
+		return this.reactiveType.equals(((ReactiveTypeDescriptor) other).reactiveType);
+	}
+
+	@Override
+	public int hashCode() {
+		return this.reactiveType.hashCode();
+	}
+
+
+	/**
+	 * Descriptor for a reactive type that can produce 0..N values.
+	 * @param type the reactive type
+	 * @param emptySupplier a supplier of an empty-value instance of the reactive type
+	 */
+	public static ReactiveTypeDescriptor multiValue(Class<?> type, Supplier<?> emptySupplier) {
+		return new ReactiveTypeDescriptor(type, emptySupplier, true, true, false);
+	}
+
+	/**
+	 * Descriptor for a reactive type that can produce 0..1 values.
+	 * @param type the reactive type
+	 * @param emptySupplier a supplier of an empty-value instance of the reactive type
+	 */
+	public static ReactiveTypeDescriptor singleOptionalValue(Class<?> type, Supplier<?> emptySupplier) {
+		return new ReactiveTypeDescriptor(type, emptySupplier, false, true, false);
+	}
+
+	/**
+	 * Descriptor for a reactive type that must produce 1 value to complete.
+	 * @param type the reactive type
+	 */
+	public static ReactiveTypeDescriptor singleRequiredValue(Class<?> type) {
+		return new ReactiveTypeDescriptor(type, null, false, false, false);
+	}
+
+	/**
+	 * Descriptor for a reactive type that does not produce any values.
+	 * @param type the reactive type
+	 * @param emptySupplier a supplier of an empty-value instance of the reactive type
+	 */
+	public static ReactiveTypeDescriptor noValue(Class<?> type, Supplier<?> emptySupplier) {
+		return new ReactiveTypeDescriptor(type, emptySupplier, false, true, true);
+	}
+
+}

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.core.convert.support;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 import io.reactivex.Flowable;
@@ -31,57 +33,211 @@ import rx.Single;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
 
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 /**
  * Unit tests for {@link ReactiveAdapterRegistry}.
  * @author Rossen Stoyanchev
  */
+@SuppressWarnings("unchecked")
 public class ReactiveAdapterRegistryTests {
 
-	private ReactiveAdapterRegistry adapterRegistry;
+	private ReactiveAdapterRegistry registry;
+
 
 	@Before
 	public void setUp() throws Exception {
-		this.adapterRegistry = new ReactiveAdapterRegistry();
+		this.registry = new ReactiveAdapterRegistry();
 	}
 
+
 	@Test
 	public void getDefaultAdapters() throws Exception {
-		testMonoAdapter(Mono.class);
-		testFluxAdapter(Flux.class);
-		testFluxAdapter(Publisher.class);
-		testMonoAdapter(CompletableFuture.class);
-		testFluxAdapter(Observable.class);
-		testMonoAdapter(Single.class);
-		testMonoAdapter(Completable.class);
-		testFluxAdapter(Flowable.class);
-		testFluxAdapter(io.reactivex.Observable.class);
-		testMonoAdapter(io.reactivex.Single.class);
-		testMonoAdapter(Maybe.class);
-		testMonoAdapter(io.reactivex.Completable.class);
+
+		// Reactor
+		assertNotNull(getAdapterTo(Mono.class));
+		assertNotNull(getAdapterTo(Flux.class));
+
+		assertNotNull(getAdapterTo(Publisher.class));
+		assertNotNull(getAdapterTo(CompletableFuture.class));
+
+		// RxJava 1
+		assertNotNull(getAdapterTo(Observable.class));
+		assertNotNull(getAdapterTo(Single.class));
+		assertNotNull(getAdapterTo(Completable.class));
+
+		// RxJava 2
+		assertNotNull(getAdapterTo(Flowable.class));
+		assertNotNull(getAdapterTo(io.reactivex.Observable.class));
+		assertNotNull(getAdapterTo(io.reactivex.Single.class));
+		assertNotNull(getAdapterTo(Maybe.class));
+		assertNotNull(getAdapterTo(io.reactivex.Completable.class));
 	}
 
-	private void testFluxAdapter(Class<?> adapteeType) {
-		ReactiveAdapter adapter = this.adapterRegistry.getAdapterFrom(adapteeType);
-		assertNotNull(adapter);
-		assertTrue(adapter.getDescriptor().isMultiValue());
-
-		adapter = this.adapterRegistry.getAdapterTo(adapteeType);
-		assertNotNull(adapter);
-		assertTrue(adapter.getDescriptor().isMultiValue());
+	@Test
+	public void publisherToFlux() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Publisher<Integer> source = Flowable.fromIterable(sequence);
+		Object target = getAdapterTo(Flux.class).fromPublisher(source);
+		assertTrue(target instanceof Flux);
+		assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
 	}
 
-	private void testMonoAdapter(Class<?> adapteeType) {
-		ReactiveAdapter adapter = this.adapterRegistry.getAdapterFrom(adapteeType);
-		assertNotNull(adapter);
-		assertFalse(adapter.getDescriptor().isMultiValue());
+	// TODO: publisherToMono/CompletableFuture vs Single (ISE on multiple elements)?
 
-		adapter = this.adapterRegistry.getAdapterTo(adapteeType);
-		assertNotNull(adapter);
-		assertFalse(adapter.getDescriptor().isMultiValue());
+	@Test
+	public void publisherToMono() throws Exception {
+		Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
+		Object target = getAdapterTo(Mono.class).fromPublisher(source);
+		assertTrue(target instanceof Mono);
+		assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
+	}
+
+	@Test
+	public void publisherToCompletableFuture() throws Exception {
+		Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
+		Object target = getAdapterTo(CompletableFuture.class).fromPublisher(source);
+		assertTrue(target instanceof CompletableFuture);
+		assertEquals(new Integer(1), ((CompletableFuture<Integer>) target).get());
+	}
+
+	@Test
+	public void publisherToRxObservable() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Publisher<Integer> source = Flowable.fromIterable(sequence);
+		Object target = getAdapterTo(rx.Observable.class).fromPublisher(source);
+		assertTrue(target instanceof rx.Observable);
+		assertEquals(sequence, ((rx.Observable) target).toList().toBlocking().first());
+	}
+
+	@Test
+	public void publisherToRxSingle() throws Exception {
+		Publisher<Integer> source = Flowable.fromArray(1);
+		Object target = getAdapterTo(rx.Single.class).fromPublisher(source);
+		assertTrue(target instanceof rx.Single);
+		assertEquals(new Integer(1), ((rx.Single<Integer>) target).toBlocking().value());
+	}
+
+	@Test
+	public void publisherToRxCompletable() throws Exception {
+		Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
+		Object target = getAdapterTo(rx.Completable.class).fromPublisher(source);
+		assertTrue(target instanceof rx.Completable);
+		assertNull(((rx.Completable) target).get());
+	}
+
+	@Test
+	public void publisherToReactivexFlowable() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Publisher<Integer> source = Flux.fromIterable(sequence);
+		Object target = getAdapterTo(io.reactivex.Flowable.class).fromPublisher(source);
+		assertTrue(target instanceof io.reactivex.Flowable);
+		assertEquals(sequence, ((io.reactivex.Flowable) target).toList().blockingGet());
+	}
+
+	@Test
+	public void publisherToReactivexObservable() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Publisher<Integer> source = Flowable.fromIterable(sequence);
+		Object target = getAdapterTo(io.reactivex.Observable.class).fromPublisher(source);
+		assertTrue(target instanceof io.reactivex.Observable);
+		assertEquals(sequence, ((io.reactivex.Observable) target).toList().blockingGet());
+	}
+
+	@Test
+	public void publisherToReactivexSingle() throws Exception {
+		Publisher<Integer> source = Flowable.fromArray(1);
+		Object target = getAdapterTo(io.reactivex.Single.class).fromPublisher(source);
+		assertTrue(target instanceof io.reactivex.Single);
+		assertEquals(new Integer(1), ((io.reactivex.Single<Integer>) target).blockingGet());
+	}
+
+	@Test
+	public void publisherToReactivexCompletable() throws Exception {
+		Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
+		Object target = getAdapterTo(io.reactivex.Completable.class).fromPublisher(source);
+		assertTrue(target instanceof io.reactivex.Completable);
+		assertNull(((io.reactivex.Completable) target).blockingGet());
+	}
+
+	@Test
+	public void rxObservableToPublisher() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Object source = rx.Observable.from(sequence);
+		Object target = getAdapterFrom(rx.Observable.class).toPublisher(source);
+		assertTrue("Expected Flux Publisher: " + target.getClass().getName(), target instanceof Flux);
+		assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
+	}
+
+	@Test
+	public void rxSingleToPublisher() throws Exception {
+		Object source = rx.Single.just(1);
+		Object target = getAdapterFrom(rx.Single.class).toPublisher(source);
+		assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
+		assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
+	}
+
+	@Test
+	public void rxCompletableToPublisher() throws Exception {
+		Object source = rx.Completable.complete();
+		Object target = getAdapterFrom(rx.Completable.class).toPublisher(source);
+		assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
+		((Mono<Void>) target).blockMillis(1000);
+	}
+
+	@Test
+	public void reactivexFlowableToPublisher() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Object source = io.reactivex.Flowable.fromIterable(sequence);
+		Object target = getAdapterFrom(io.reactivex.Flowable.class).toPublisher(source);
+		assertTrue("Expected Flux Publisher: " + target.getClass().getName(), target instanceof Flux);
+		assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
+	}
+
+	@Test
+	public void reactivexObservableToPublisher() throws Exception {
+		List<Integer> sequence = Arrays.asList(1, 2, 3);
+		Object source = io.reactivex.Observable.fromIterable(sequence);
+		Object target = getAdapterFrom(io.reactivex.Observable.class).toPublisher(source);
+		assertTrue("Expected Flux Publisher: " + target.getClass().getName(), target instanceof Flux);
+		assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
+	}
+
+	@Test
+	public void reactivexSingleToPublisher() throws Exception {
+		Object source = io.reactivex.Single.just(1);
+		Object target = getAdapterFrom(io.reactivex.Single.class).toPublisher(source);
+		assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
+		assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
+	}
+
+	@Test
+	public void reactivexCompletableToPublisher() throws Exception {
+		Object source = io.reactivex.Completable.complete();
+		Object target = getAdapterFrom(io.reactivex.Completable.class).toPublisher(source);
+		assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
+		((Mono<Void>) target).blockMillis(1000);
+	}
+
+	@Test
+	public void CompletableFutureToPublisher() throws Exception {
+		CompletableFuture<Integer> future = new CompletableFuture();
+		future.complete(1);
+		Object target = getAdapterFrom(CompletableFuture.class).toPublisher(future);
+		assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
+		assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
+	}
+
+
+	private ReactiveAdapter getAdapterTo(Class<?> reactiveType) {
+		return this.registry.getAdapterTo(reactiveType);
+	}
+
+	private ReactiveAdapter getAdapterFrom(Class<?> reactiveType) {
+		return this.registry.getAdapterFrom(reactiveType);
 	}
 
 }

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/BindingContextFactory.java

@@ -149,7 +149,7 @@ class BindingContextFactory {
 		Class<?> valueType = (adapter != null ? type.resolveGeneric(0) : type.resolve());
 
 		if (Void.class.equals(valueType) || void.class.equals(valueType)) {
-			return (adapter != null ? adapter.toMono(value) : Mono.empty());
+			return (adapter != null ? Mono.from(adapter.toPublisher(value)) : Mono.empty());
 		}
 
 		String name = getAttributeName(valueType, result.getReturnTypeSource());

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ModelAttributeMethodArgumentResolver.java

@@ -25,8 +25,8 @@ import reactor.core.publisher.MonoProcessor;
 import org.springframework.beans.BeanUtils;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ReactiveAdapter;
-import org.springframework.core.ReactiveAdapter.Descriptor;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.ReactiveTypeDescriptor;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.util.Assert;
@@ -107,7 +107,7 @@ public class ModelAttributeMethodArgumentResolver implements HandlerMethodArgume
 			Class<?> clazz = parameter.getParameterType();
 			ReactiveAdapter adapter = getAdapterRegistry().getAdapterFrom(clazz);
 			if (adapter != null) {
-				Descriptor descriptor = adapter.getDescriptor();
+				ReactiveTypeDescriptor descriptor = adapter.getDescriptor();
 				if (descriptor.isNoValue() || descriptor.isMultiValue()) {
 					return false;
 				}
@@ -179,12 +179,15 @@ public class ModelAttributeMethodArgumentResolver implements HandlerMethodArgume
 		if (attribute != null) {
 			ReactiveAdapter adapterFrom = getAdapterRegistry().getAdapterFrom(null, attribute);
 			if (adapterFrom != null) {
-				return adapterFrom.toMono(attribute);
+				ReactiveTypeDescriptor descriptor = adapterFrom.getDescriptor();
+				Assert.isTrue(!descriptor.isMultiValue(), "Data binding supports single-value async types.");
+				return Mono.from(adapterFrom.toPublisher(attribute));
 			}
 		}
 		return Mono.justOrEmpty(attribute);
 	}
 
+
 	protected Object createAttribute(String attributeName, Class<?> attributeType,
 			MethodParameter parameter, BindingContext context, ServerWebExchange exchange) {
 

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/method/annotation/ResponseEntityResultHandler.java

@@ -25,6 +25,7 @@ import reactor.core.publisher.Mono;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.ReactiveTypeDescriptor;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
@@ -120,7 +121,9 @@ public class ResponseEntityResultHandler extends AbstractMessageWriterResultHand
 		ReactiveAdapter adapter = getAdapterRegistry().getAdapterFrom(rawClass, optionalValue);
 
 		if (adapter != null) {
-			returnValueMono = adapter.toMono(optionalValue);
+			ReactiveTypeDescriptor descriptor = adapter.getDescriptor();
+			Assert.isTrue(!descriptor.isMultiValue(), "Only a single ResponseEntity supported.");
+			returnValueMono = Mono.from(adapter.toPublisher(optionalValue));
 			bodyType = new MethodParameter(result.getReturnTypeSource());
 			bodyType.increaseNestingLevel();
 			bodyType.increaseNestingLevel();

spring-web-reactive/src/main/java/org/springframework/web/reactive/result/view/ViewResolutionResultHandler.java

@@ -33,10 +33,12 @@ import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.ReactiveTypeDescriptor;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 import org.springframework.http.MediaType;
 import org.springframework.ui.Model;
+import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.validation.BindingResult;
@@ -191,8 +193,10 @@ public class ViewResolutionResultHandler extends AbstractHandlerResultHandler
 		ReactiveAdapter adapter = getAdapterRegistry().getAdapterFrom(parameterType.getRawClass(), optional);
 
 		if (adapter != null) {
+			ReactiveTypeDescriptor descriptor = adapter.getDescriptor();
+			Assert.isTrue(!descriptor.isMultiValue(), "Only single-value async return type supported.");
 			returnValueMono = optional
-					.map(value -> adapter.toMono(value).cast(Object.class))
+					.map(value -> Mono.from(adapter.toPublisher(value)))
 					.orElse(Mono.empty());
 			elementType = !adapter.getDescriptor().isNoValue() ?
 					parameterType.getGeneric(0) : ResolvableType.forClass(Void.class);
@@ -301,11 +305,11 @@ public class ViewResolutionResultHandler extends AbstractHandlerResultHandler
 			if (adapter != null) {
 				names.add(entry.getKey());
 				if (adapter.getDescriptor().isMultiValue()) {
-					Flux<Object> value = adapter.toFlux(entry.getValue());
+					Flux<Object> value = Flux.from(adapter.toPublisher(entry.getValue()));
 					valueMonos.add(value.collectList().defaultIfEmpty(Collections.emptyList()));
 				}
 				else {
-					Mono<Object> value = adapter.toMono(entry.getValue());
+					Mono<Object> value = Mono.from(adapter.toPublisher(entry.getValue()));
 					valueMonos.add(value.defaultIfEmpty(NO_VALUE));
 				}
 			}