Polish ReactiveAdapterRegisry

Author: rstoyanchev

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -24,11 +24,10 @@
 import org.springframework.util.Assert;
 
 /**
- * Adapt a Reactive Streams {@link Publisher} to and from an async/reactive type
- * such as {@code CompletableFuture}, an RxJava {@code Observable}, etc.
+ * Adapter for a Reactive Streams {@link Publisher} to and from an async/reactive
+ * type such as {@code CompletableFuture}, RxJava {@code Observable}, and others.
  *
- * <p>Use the {@link ReactiveAdapterRegistry} to register reactive types and
- * obtain adapters from.
+ * <p>An adapter is typically obtained via {@link ReactiveAdapterRegistry}.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
@@ -71,37 +70,38 @@ public ReactiveTypeDescriptor getDescriptor() {
 	}
 
 	/**
-	 * A shortcut for {@code getDescriptor().getReactiveType()}.
+	 * Shortcut for {@code getDescriptor().getReactiveType()}.
 	 */
 	public Class<?> getReactiveType() {
 		return getDescriptor().getReactiveType();
 	}
 
 	/**
-	 * A shortcut for {@code getDescriptor().isMultiValue()}.
+	 * Shortcut for {@code getDescriptor().isMultiValue()}.
 	 */
 	public boolean isMultiValue() {
 		return getDescriptor().isMultiValue();
 	}
 
 	/**
-	 * A shortcut for {@code getDescriptor().supportsEmpty()}.
+	 * Shortcut for {@code getDescriptor().supportsEmpty()}.
 	 */
 	public boolean supportsEmpty() {
 		return getDescriptor().supportsEmpty();
 	}
 
 	/**
-	 * A shortcut for {@code getDescriptor().isNoValue()}.
+	 * Shortcut for {@code getDescriptor().isNoValue()}.
 	 */
 	public boolean isNoValue() {
 		return getDescriptor().isNoValue();
 	}
 
 
 	/**
-	 * Adapt the given instance to a Reactive Streams Publisher.
-	 * @param source the source object to adapt from
+	 * Adapt the given instance to a Reactive Streams {@code Publisher}.
+	 * @param source the source object to adapt from; if the given object is
+	 * {@code null}, {@link ReactiveTypeDescriptor#getEmptyValue()} is used.
 	 * @return the Publisher representing the adaptation
 	 */
 	@SuppressWarnings("unchecked")

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -34,13 +34,8 @@
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ReflectionUtils;
 
-import static org.springframework.core.ReactiveTypeDescriptor.multiValue;
-import static org.springframework.core.ReactiveTypeDescriptor.noValue;
-import static org.springframework.core.ReactiveTypeDescriptor.singleOptionalValue;
-import static org.springframework.core.ReactiveTypeDescriptor.singleRequiredValue;
-
 /**
- * A registry of adapters to adapt a Reactive Streams {@link Publisher} to/from
+ * A registry of adapters to adapt Reactive Streams {@link Publisher} to/from
  * various async/reactive types such as {@code CompletableFuture}, RxJava
  * {@code Observable}, and others.
  *
@@ -64,6 +59,7 @@ public class ReactiveAdapterRegistry {
 
 	/**
 	 * Create a registry and auto-register default adapters.
+	 * @see #getSharedInstance()
 	 */
 	public ReactiveAdapterRegistry() {
 
@@ -168,49 +164,54 @@ public ReactiveAdapter getAdapter(@Nullable Class<?> reactiveType, @Nullable Obj
 	/**
 	 * Return a shared default {@code ReactiveAdapterRegistry} instance, lazily
 	 * building it once needed.
+	 *
 	 * <p><b>NOTE:</b> We highly recommend passing a long-lived, pre-configured
 	 * {@code ReactiveAdapterRegistry} instance for customization purposes.
 	 * This accessor is only meant as a fallback for code paths that want to
 	 * fall back on a default instance if one isn't provided.
+	 *
 	 * @return the shared {@code ReactiveAdapterRegistry} instance (never {@code null})
 	 * @since 5.0.2
 	 */
 	public static ReactiveAdapterRegistry getSharedInstance() {
-		ReactiveAdapterRegistry ar = sharedInstance;
-		if (ar == null) {
+		ReactiveAdapterRegistry registry = sharedInstance;
+		if (registry == null) {
 			synchronized (ReactiveAdapterRegistry.class) {
-				ar = sharedInstance;
-				if (ar == null) {
-					ar = new ReactiveAdapterRegistry();
-					sharedInstance = ar;
+				registry = sharedInstance;
+				if (registry == null) {
+					registry = new ReactiveAdapterRegistry();
+					sharedInstance = registry;
 				}
 			}
 		}
-		return ar;
+		return registry;
 	}
 
 
 	private static class ReactorRegistrar {
 
 		void registerAdapters(ReactiveAdapterRegistry registry) {
-			// Flux and Mono ahead of Publisher...
+
+			// Register Flux and Mono before Publisher...
 
 			registry.registerReactiveType(
-					singleOptionalValue(Mono.class, Mono::empty),
+					ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
 					source -> (Mono<?>) source,
 					Mono::from
 			);
 
-			registry.registerReactiveType(multiValue(Flux.class, Flux::empty),
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
 					source -> (Flux<?>) source,
 					Flux::from);
 
-			registry.registerReactiveType(multiValue(Publisher.class, Flux::empty),
+			registry.registerReactiveType(
+					ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty),
 					source -> (Publisher<?>) source,
 					source -> source);
 
 			registry.registerReactiveType(
-					singleOptionalValue(CompletableFuture.class, () -> {
+					ReactiveTypeDescriptor.singleOptionalValue(CompletableFuture.class, () -> {
 						CompletableFuture<?> empty = new CompletableFuture<>();
 						empty.complete(null);
 						return empty;
@@ -226,17 +227,17 @@ private static class RxJava1Registrar {
 
 		void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
-					multiValue(rx.Observable.class, rx.Observable::empty),
+					ReactiveTypeDescriptor.multiValue(rx.Observable.class, rx.Observable::empty),
 					source -> RxReactiveStreams.toPublisher((rx.Observable<?>) source),
 					RxReactiveStreams::toObservable
 			);
 			registry.registerReactiveType(
-					singleRequiredValue(rx.Single.class),
+					ReactiveTypeDescriptor.singleRequiredValue(rx.Single.class),
 					source -> RxReactiveStreams.toPublisher((rx.Single<?>) source),
 					RxReactiveStreams::toSingle
 			);
 			registry.registerReactiveType(
-					noValue(rx.Completable.class, rx.Completable::complete),
+					ReactiveTypeDescriptor.noValue(rx.Completable.class, rx.Completable::complete),
 					source -> RxReactiveStreams.toPublisher((rx.Completable) source),
 					RxReactiveStreams::toCompletable
 			);
@@ -248,27 +249,27 @@ private static class RxJava2Registrar {
 
 		void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
-					multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
+					ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
 					source -> (io.reactivex.Flowable<?>) source,
 					Flowable::fromPublisher
 			);
 			registry.registerReactiveType(
-					multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
+					ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
 					source -> ((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
 					source -> io.reactivex.Flowable.fromPublisher(source).toObservable()
 			);
 			registry.registerReactiveType(
-					singleRequiredValue(io.reactivex.Single.class),
+					ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
 					source -> ((io.reactivex.Single<?>) source).toFlowable(),
 					source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
 			);
 			registry.registerReactiveType(
-					singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
+					ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
 					source -> ((io.reactivex.Maybe<?>) source).toFlowable(),
 					source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement()
 			);
 			registry.registerReactiveType(
-					noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
+					ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
 					source -> ((io.reactivex.Completable) source).toFlowable(),
 					source -> io.reactivex.Flowable.fromPublisher(source).toObservable().ignoreElements()
 			);
@@ -278,15 +279,15 @@ void registerAdapters(ReactiveAdapterRegistry registry) {
 
 	private static class ReactorJdkFlowAdapterRegistrar {
 
-		// TODO: remove reflection when build requires JDK 9+
 		void registerAdapter(ReactiveAdapterRegistry registry) throws Exception {
+			// TODO: remove reflection when build requires JDK 9+
 			Class<?> type = ClassUtils.forName("java.util.concurrent.Flow.Publisher", getClass().getClassLoader());
 			Method toFluxMethod = getMethod("flowPublisherToFlux", type);
 			Method toFlowMethod = getMethod("publisherToFlowPublisher", Publisher.class);
 			Object emptyFlow = ReflectionUtils.invokeMethod(toFlowMethod, null, Flux.empty());
 
 			registry.registerReactiveType(
-					multiValue(type, () -> emptyFlow),
+					ReactiveTypeDescriptor.multiValue(type, () -> emptyFlow),
 					source -> (Publisher<?>) ReflectionUtils.invokeMethod(toFluxMethod, null, source),
 					publisher -> ReflectionUtils.invokeMethod(toFlowMethod, null, publisher)
 			);
@@ -299,9 +300,10 @@ private static Method getMethod(String name, Class<?> argumentType) throws NoSuc
 
 
 	/**
-	 * Extension of ReactiveAdapter that wraps adapted (raw) Publisher's as
-	 * {@link Flux} or {@link Mono} depending on the underlying reactive type's
-	 * stream semantics.
+	 * 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 {
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -22,8 +22,8 @@
 import org.springframework.util.Assert;
 
 /**
- * Descriptor for a reactive type with information about its stream semantics, i.e.
- * how many values it can produce.
+ * Describes the semantics of a reactive type including boolean checks for
+ * {@link #isMultiValue()}, {@link #supportsEmpty()}, and {@link #isNoValue()}.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
@@ -55,21 +55,12 @@ private ReactiveTypeDescriptor(Class<?> reactiveType, @Nullable Supplier<?> empt
 
 
 	/**
-	 * Return the reactive type the descriptor was created for.
+	 * Return the reactive type for this descriptor.
 	 */
 	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.state(this.emptyValueSupplier != null, "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}.
@@ -95,6 +86,15 @@ public boolean isNoValue() {
 		return this.noValue;
 	}
 
+	/**
+	 * 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.state(this.emptyValueSupplier != null, "Empty values not supported");
+		return this.emptyValueSupplier.get();
+	}
+
 
 	@Override
 	public boolean equals(@Nullable Object other) {