DATACMNS-947 - Adopt ReactiveTypeDescriptor and ReactiveAdapterRegistry changes.

Remove our own ReactiveWrappers.Descriptor type in favor of ReactiveTypeDescriptor. Adopt changes in ReactiveAdapterRegistry.

Author: mp911de

src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java

@@ -169,7 +169,7 @@ private static ConversionService registerConvertersIn(ConfigurableConversionServ
 	 */
 	public static boolean supports(Class<?> type) {
 		return RegistryHolder.REACTIVE_ADAPTER_REGISTRY != null
-				&& RegistryHolder.REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(type) != null;
+				&& RegistryHolder.REACTIVE_ADAPTER_REGISTRY.getAdapter(type) != null;
 	}
 
 	/**
@@ -476,7 +476,7 @@ private enum PublisherToRxJava1SingleConverter implements Converter<Publisher<?>
 
 		@Override
 		public Single<?> convert(Publisher<?> source) {
-			return (Single<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Single.class).fromPublisher(Mono.from(source));
+			return (Single<?>) REACTIVE_ADAPTER_REGISTRY.getAdapter(Single.class).fromPublisher(Mono.from(source));
 		}
 	}
 
@@ -492,7 +492,7 @@ private enum PublisherToRxJava1CompletableConverter implements Converter<Publish
 
 		@Override
 		public Completable convert(Publisher<?> source) {
-			return (Completable) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Completable.class).fromPublisher(source);
+			return (Completable) REACTIVE_ADAPTER_REGISTRY.getAdapter(Completable.class).fromPublisher(source);
 		}
 	}
 
@@ -508,7 +508,7 @@ private enum PublisherToRxJava1ObservableConverter implements Converter<Publishe
 
 		@Override
 		public Observable<?> convert(Publisher<?> source) {
-			return (Observable<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Observable.class).fromPublisher(Flux.from(source));
+			return (Observable<?>) REACTIVE_ADAPTER_REGISTRY.getAdapter(Observable.class).fromPublisher(Flux.from(source));
 		}
 	}
 
@@ -524,7 +524,7 @@ private enum RxJava1SingleToPublisherConverter implements Converter<Single<?>, P
 
 		@Override
 		public Publisher<?> convert(Single<?> source) {
-			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toPublisher(source));
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapter(Single.class).toPublisher(source));
 		}
 	}
 
@@ -540,7 +540,7 @@ private enum RxJava1SingleToMonoConverter implements Converter<Single<?>, Mono<?
 
 		@Override
 		public Mono<?> convert(Single<?> source) {
-			return Mono.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toMono(source));
+			return Mono.defer(() -> Mono.from(REACTIVE_ADAPTER_REGISTRY.getAdapter(Single.class).toPublisher(source)));
 		}
 	}
 
@@ -556,7 +556,7 @@ private enum RxJava1SingleToFluxConverter implements Converter<Single<?>, Flux<?
 
 		@Override
 		public Flux<?> convert(Single<?> source) {
-			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toFlux(source));
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapter(Single.class).toPublisher(source));
 		}
 	}
 
@@ -572,7 +572,7 @@ private enum RxJava1CompletableToPublisherConverter implements Converter<Complet
 
 		@Override
 		public Publisher<?> convert(Completable source) {
-			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Completable.class).toFlux(source));
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapter(Completable.class).toPublisher(source));
 		}
 	}
 
@@ -604,7 +604,7 @@ private enum RxJava1ObservableToPublisherConverter implements Converter<Observab
 
 		@Override
 		public Publisher<?> convert(Observable<?> source) {
-			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source));
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapter(Observable.class).toPublisher(source));
 		}
 	}
 
@@ -620,7 +620,7 @@ private enum RxJava1ObservableToMonoConverter implements Converter<Observable<?>
 
 		@Override
 		public Mono<?> convert(Observable<?> source) {
-			return Mono.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toMono(source));
+			return Mono.defer(() -> Mono.from(REACTIVE_ADAPTER_REGISTRY.getAdapter(Observable.class).toPublisher(source)));
 		}
 	}
 
@@ -636,7 +636,7 @@ private enum RxJava1ObservableToFluxConverter implements Converter<Observable<?>
 
 		@Override
 		public Flux<?> convert(Observable<?> source) {
-			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source));
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapter(Observable.class).toPublisher(source));
 		}
 	}
 
@@ -688,7 +688,7 @@ private enum PublisherToRxJava2SingleConverter implements Converter<Publisher<?>
 
 		@Override
 		public io.reactivex.Single<?> convert(Publisher<?> source) {
-			return (io.reactivex.Single<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(io.reactivex.Single.class)
+			return (io.reactivex.Single<?>) REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Single.class)
 					.fromPublisher(source);
 		}
 	}
@@ -705,7 +705,7 @@ private enum PublisherToRxJava2CompletableConverter implements Converter<Publish
 
 		@Override
 		public io.reactivex.Completable convert(Publisher<?> source) {
-			return (io.reactivex.Completable) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(io.reactivex.Completable.class)
+			return (io.reactivex.Completable) REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Completable.class)
 					.fromPublisher(source);
 		}
 	}
@@ -722,7 +722,7 @@ private enum PublisherToRxJava2ObservableConverter implements Converter<Publishe
 
 		@Override
 		public io.reactivex.Observable<?> convert(Publisher<?> source) {
-			return (io.reactivex.Observable<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(io.reactivex.Single.class)
+			return (io.reactivex.Observable<?>) REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Single.class)
 					.fromPublisher(source);
 		}
 	}
@@ -739,7 +739,7 @@ private enum RxJava2SingleToPublisherConverter implements Converter<io.reactivex
 
 		@Override
 		public Publisher<?> convert(io.reactivex.Single<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Single.class).toMono(source);
+			return REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Single.class).toPublisher(source);
 		}
 	}
 
@@ -755,7 +755,7 @@ private enum RxJava2SingleToMonoConverter implements Converter<io.reactivex.Sing
 
 		@Override
 		public Mono<?> convert(io.reactivex.Single<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Single.class).toMono(source);
+			return Mono.from(REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Single.class).toPublisher(source));
 		}
 	}
 
@@ -771,7 +771,7 @@ private enum RxJava2SingleToFluxConverter implements Converter<io.reactivex.Sing
 
 		@Override
 		public Flux<?> convert(io.reactivex.Single<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Single.class).toFlux(source);
+			return Flux.from(REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Single.class).toPublisher(source));
 		}
 	}
 
@@ -787,7 +787,7 @@ private enum RxJava2CompletableToPublisherConverter implements Converter<io.reac
 
 		@Override
 		public Publisher<?> convert(io.reactivex.Completable source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Completable.class).toFlux(source);
+			return REACTIVE_ADAPTER_REGISTRY.getAdapter(io.reactivex.Completable.class).toPublisher(source);
 		}
 	}
 
@@ -899,7 +899,7 @@ private enum PublisherToRxJava2MaybeConverter implements Converter<Publisher<?>,
 
 		@Override
 		public io.reactivex.Maybe<?> convert(Publisher<?> source) {
-			return (io.reactivex.Maybe<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Maybe.class).fromPublisher(source);
+			return (io.reactivex.Maybe<?>) REACTIVE_ADAPTER_REGISTRY.getAdapter(Maybe.class).fromPublisher(source);
 		}
 	}
 

src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java

@@ -15,27 +15,23 @@
  */
 package org.springframework.data.repository.util;
 
-import lombok.AccessLevel;
-import lombok.RequiredArgsConstructor;
-import lombok.Value;
 import lombok.experimental.UtilityClass;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import rx.Completable;
 import rx.Observable;
 import rx.Single;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
-import org.springframework.core.ReactiveAdapter;
+
+import org.springframework.core.ReactiveTypeDescriptor;
 import org.springframework.data.util.ReflectionUtils;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -75,7 +71,7 @@ public class ReactiveWrappers {
 	private static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Flowable",
 			ReactiveWrappers.class.getClassLoader());
 
-	private static final Map<Class<?>, Descriptor> REACTIVE_WRAPPERS;
+	private static final Collection<ReactiveTypeDescriptor> REACTIVE_WRAPPERS;
 
 	/**
 	 * Enumeration of supported reactive libraries.
@@ -88,32 +84,36 @@ static enum ReactiveLibrary {
 
 	static {
 
-		Map<Class<?>, Descriptor> reactiveWrappers = new LinkedHashMap<>(5);
+		Collection<ReactiveTypeDescriptor> reactiveWrappers = new ArrayList<>(5);
 
 		if (RXJAVA1_PRESENT) {
 
-			reactiveWrappers.put(Single.class, Descriptor.forSingleValue().forValue());
-			reactiveWrappers.put(Completable.class, Descriptor.forSingleValue().forNoValue());
-			reactiveWrappers.put(Observable.class, Descriptor.forMultiValue().forValue());
+			reactiveWrappers.add(ReactiveTypeDescriptor.singleRequiredValue(Single.class));
+			reactiveWrappers.add(ReactiveTypeDescriptor.noValue(Completable.class, Completable::complete));
+			reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Observable.class, Observable::empty));
 		}
 
 		if (RXJAVA2_PRESENT) {
 
-			reactiveWrappers.put(io.reactivex.Single.class, Descriptor.forSingleValue().forValue());
-			reactiveWrappers.put(io.reactivex.Maybe.class, Descriptor.forSingleValue().forValue());
-			reactiveWrappers.put(io.reactivex.Completable.class, Descriptor.forSingleValue().forNoValue());
-			reactiveWrappers.put(io.reactivex.Flowable.class, Descriptor.forMultiValue().forValue());
-			reactiveWrappers.put(io.reactivex.Observable.class, Descriptor.forMultiValue().forValue());
+			reactiveWrappers.add(ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class));
+			reactiveWrappers
+					.add(ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty));
+			reactiveWrappers
+					.add(ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete));
+			reactiveWrappers
+					.add(ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty));
+			reactiveWrappers
+					.add(ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty));
 		}
 
 		if (PROJECT_REACTOR_PRESENT) {
 
-			reactiveWrappers.put(Mono.class, Descriptor.forSingleValue().forValue());
-			reactiveWrappers.put(Flux.class, Descriptor.forMultiValue().forNoValue());
-			reactiveWrappers.put(Publisher.class, Descriptor.forMultiValue().forNoValue());
+			reactiveWrappers.add(ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty));
+			reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty));
+			reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty));
 		}
 
-		REACTIVE_WRAPPERS = Collections.unmodifiableMap(reactiveWrappers);
+		REACTIVE_WRAPPERS = Collections.unmodifiableCollection(reactiveWrappers);
 	}
 
 	/**
@@ -183,7 +183,7 @@ public static boolean isNoValueType(Class<?> type) {
 
 		Assert.notNull(type, "Candidate type must not be null!");
 
-		return findDescriptor(type).map(Descriptor::isNoValue).orElse(false);
+		return findDescriptor(type).map(ReactiveTypeDescriptor::isNoValue).orElse(false);
 	}
 
 	/**
@@ -213,7 +213,8 @@ public static boolean isMultiValueType(Class<?> type) {
 
 		// Prevent single-types with a multi-hierarchy supertype to be reported as multi type
 		// See Mono implements Publisher
-		return isSingleValueType(type) ? false : findDescriptor(type).map(Descriptor::isMultiValue).orElse(false);
+		return isSingleValueType(type) ? false
+				: findDescriptor(type).map(ReactiveTypeDescriptor::isMultiValue).orElse(false);
 	}
 
 	/**
@@ -223,9 +224,10 @@ public static boolean isMultiValueType(Class<?> type) {
 	 */
 	public static Collection<Class<?>> getNoValueTypes() {
 
-		return REACTIVE_WRAPPERS.entrySet().stream()//
-				.filter(entry -> entry.getValue().isNoValue())//
-				.map(Entry::getKey).collect(Collectors.toList());
+		return REACTIVE_WRAPPERS.stream()//
+				.filter(ReactiveTypeDescriptor::isNoValue)//
+				.map(ReactiveTypeDescriptor::getReactiveType)//
+				.collect(Collectors.toList());
 	}
 
 	/**
@@ -235,9 +237,9 @@ public static Collection<Class<?>> getNoValueTypes() {
 	 */
 	public static Collection<Class<?>> getSingleValueTypes() {
 
-		return REACTIVE_WRAPPERS.entrySet().stream()//
-				.filter(entry -> !entry.getValue().isMultiValue())//
-				.map(Entry::getKey).collect(Collectors.toList());
+		return REACTIVE_WRAPPERS.stream()//
+				.filter(entry -> !entry.isMultiValue())//
+				.map(ReactiveTypeDescriptor::getReactiveType).collect(Collectors.toList());
 	}
 
 	/**
@@ -247,9 +249,9 @@ public static Collection<Class<?>> getSingleValueTypes() {
 	 */
 	public static Collection<Class<?>> getMultiValueTypes() {
 
-		return REACTIVE_WRAPPERS.entrySet().stream()//
-				.filter(entry -> entry.getValue().isMultiValue())//
-				.map(Entry::getKey)//
+		return REACTIVE_WRAPPERS.stream()//
+				.filter(ReactiveTypeDescriptor::isMultiValue)//
+				.map(ReactiveTypeDescriptor::getReactiveType)//
 				.collect(Collectors.toList());
 	}
 
@@ -267,64 +269,17 @@ private static boolean isWrapper(Class<?> type) {
 	}
 
 	/**
-	 * Looks up a {@link Descriptor} for the given wrapper type.
+	 * Looks up a {@link ReactiveTypeDescriptor} for the given wrapper type.
 	 * 
 	 * @param type must not be {@literal null}.
 	 * @return
 	 */
-	private static Optional<Descriptor> findDescriptor(Class<?> type) {
+	private static Optional<ReactiveTypeDescriptor> findDescriptor(Class<?> type) {
 
 		Assert.notNull(type, "Wrapper type must not be null!");
 
-		return REACTIVE_WRAPPERS.entrySet().stream()//
-				.filter(it -> ClassUtils.isAssignable(it.getKey(), type))//
-				.findFirst().map(it -> it.getValue());
-	}
-
-	/**
-	 * Basically a copy of Spring's {@link ReactiveAdapter.Descriptor} but without introducing the strong dependency to
-	 * Reactor so that we can safely use the class in non-reactive environments.
-	 *
-	 * @author Oliver Gierke
-	 * @since 2.0
-	 */
-	@Value
-	@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
-	private static class Descriptor {
-
-		/**
-		 * 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}.
-		 */
-		private final boolean multiValue;
-
-		/**
-		 * Return {@code true} if the adaptee implies no values will be produced, i.e. providing only completion or error
-		 * signal.
-		 */
-		private final boolean noValue;
-
-		public static DescriptorBuilder forSingleValue() {
-			return new DescriptorBuilder(false);
-		}
-
-		public static DescriptorBuilder forMultiValue() {
-			return new DescriptorBuilder(true);
-		}
-
-		@RequiredArgsConstructor
-		static class DescriptorBuilder {
-
-			private final boolean multi;
-
-			public Descriptor forValue() {
-				return new Descriptor(multi, false);
-			}
-
-			public Descriptor forNoValue() {
-				return new Descriptor(multi, true);
-			}
-		}
+		return REACTIVE_WRAPPERS.stream()//
+				.filter(it -> ClassUtils.isAssignable(it.getReactiveType(), type))//
+				.findFirst();
 	}
 }