Spring/spring-data-commons
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACMNS-949 - Drop own converters in favor of ReactiveAdapterRegistry.

Browse Source 
We now reuse ReactiveAdapterRegistry as much as possible instead of registering our own converters.

Original pull request: #415.
This commit is contained in:
Mark Paluch
2019-11-14 14:35:21 +01:00
parent ce7683c672
commit c945fa4f9a
3 changed files with 37 additions and 740 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/main/java/org/springframework/data/repository/util/QueryExecutionConverters.java
View File

@@ -35,7 +35,6 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.annotation.Nonnull;
@@ -152,15 +151,6 @@ public abstract class QueryExecutionConverters {
ALLOWED_PAGEABLE_TYPES.add(io.vavr.collection.Seq.class);
}
if (ReactiveWrappers.isAvailable()) {
WRAPPER_TYPES
.addAll(ReactiveWrappers.getNoValueTypes().stream().map(WrapperType::noValue).collect(Collectors.toList()));
WRAPPER_TYPES.addAll(
ReactiveWrappers.getSingleValueTypes().stream().map(WrapperType::singleValue).collect(Collectors.toList()));
WRAPPER_TYPES.addAll(
ReactiveWrappers.getMultiValueTypes().stream().map(WrapperType::multiValue).collect(Collectors.toList()));
}
}
private QueryExecutionConverters() {}
@@ -183,6 +173,10 @@ public abstract class QueryExecutionConverters {
}
}
if (ReactiveWrappers.supports(type)) {
return true;
}
return false;
});
}
@@ -214,6 +208,10 @@ public abstract class QueryExecutionConverters {
}
}
if (ReactiveWrappers.supports(type) && ReactiveWrappers.isSingleValueType(type)) {
return true;
}
return false;
}

666
src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java
View File

@@ -15,9 +15,6 @@
*/
package org.springframework.data.repository.util;
import static org.springframework.data.repository.util.ReactiveWrapperConverters.RegistryHolder.*;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import kotlinx.coroutines.flow.Flow;
@@ -25,7 +22,6 @@ import kotlinx.coroutines.reactive.ReactiveFlowKt;
import lombok.experimental.UtilityClass;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Completable;
import rx.Observable;
import rx.Single;
@@ -107,67 +103,16 @@ public class ReactiveWrapperConverters {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.PROJECT_REACTOR)) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA1)) {
conversionService.addConverter(PublisherToRxJava1CompletableConverter.INSTANCE);
conversionService.addConverter(RxJava1CompletableToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava1CompletableToMonoConverter.INSTANCE);
conversionService.addConverter(PublisherToRxJava1SingleConverter.INSTANCE);
conversionService.addConverter(RxJava1SingleToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava1SingleToMonoConverter.INSTANCE);
conversionService.addConverter(RxJava1SingleToFluxConverter.INSTANCE);
conversionService.addConverter(PublisherToRxJava1ObservableConverter.INSTANCE);
conversionService.addConverter(RxJava1ObservableToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava1ObservableToMonoConverter.INSTANCE);
conversionService.addConverter(RxJava1ObservableToFluxConverter.INSTANCE);
}
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA2)) {
conversionService.addConverter(PublisherToRxJava2CompletableConverter.INSTANCE);
conversionService.addConverter(RxJava2CompletableToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava2CompletableToMonoConverter.INSTANCE);
conversionService.addConverter(PublisherToRxJava2SingleConverter.INSTANCE);
conversionService.addConverter(RxJava2SingleToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava2SingleToMonoConverter.INSTANCE);
conversionService.addConverter(RxJava2SingleToFluxConverter.INSTANCE);
conversionService.addConverter(PublisherToRxJava2ObservableConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToMonoConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToFluxConverter.INSTANCE);
conversionService.addConverter(PublisherToRxJava2FlowableConverter.INSTANCE);
conversionService.addConverter(RxJava2FlowableToPublisherConverter.INSTANCE);
conversionService.addConverter(PublisherToRxJava2MaybeConverter.INSTANCE);
conversionService.addConverter(RxJava2MaybeToPublisherConverter.INSTANCE);
conversionService.addConverter(RxJava2MaybeToMonoConverter.INSTANCE);
conversionService.addConverter(RxJava2MaybeToFluxConverter.INSTANCE);
}
conversionService.addConverter(PublisherToMonoConverter.INSTANCE);
conversionService.addConverter(PublisherToFluxConverter.INSTANCE);
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA1)) {
conversionService.addConverter(RxJava1SingleToObservableConverter.INSTANCE);
conversionService.addConverter(RxJava1ObservableToSingleConverter.INSTANCE);
}
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA2)) {
conversionService.addConverter(RxJava2SingleToObservableConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToSingleConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToMaybeConverter.INSTANCE);
}
if (ReactiveWrappers.isAvailable(ReactiveLibrary.KOTLIN_COROUTINES)) {
conversionService.addConverter(PublisherToFlowConverter.INSTANCE);
}
conversionService.addConverterFactory(ReactiveAdapterConverterFactory.INSTANCE);
if (RegistryHolder.REACTIVE_ADAPTER_REGISTRY != null) {
conversionService.addConverterFactory(ReactiveAdapterConverterFactory.INSTANCE);
}
}
return conversionService;
@@ -244,31 +189,6 @@ public class ReactiveWrapperConverters {
return GENERIC_CONVERSION_SERVICE.canConvert(sourceType, targetType);
}
/**
* Returns the {@link ReactiveAdapter} for the given type.
*
* @param type must not be {@literal null}.
* @return
* @throws IllegalStateException if no adapter registry could be found.
* @throws IllegalArgumentException if no adapter could be found for the given type.
*/
private static ReactiveAdapter getRequiredAdapter(Class<?> type) {
ReactiveAdapterRegistry registry = REACTIVE_ADAPTER_REGISTRY;
if (registry == null) {
throw new IllegalStateException("No reactive adapter registry found!");
}
ReactiveAdapter adapter = registry.getAdapter(type);
if (adapter == null) {
throw new IllegalArgumentException(String.format("Expected to find reactive adapter for %s but couldn't!", type));
}
return adapter;
}
// -------------------------------------------------------------------------
// Wrapper descriptors
// -------------------------------------------------------------------------
@@ -525,578 +445,6 @@ public class ReactiveWrapperConverters {
}
}
// -------------------------------------------------------------------------
// RxJava 1 converters
// -------------------------------------------------------------------------
/**
* A {@link Converter} to convert a {@link Publisher} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava1SingleConverter implements Converter<Publisher<?>, Single<?>> {
INSTANCE;
@Nonnull
@Override
public Single<?> convert(Publisher<?> source) {
return (Single<?>) getRequiredAdapter(Single.class).fromPublisher(Mono.from(source));
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link Completable}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava1CompletableConverter implements Converter<Publisher<?>, Completable> {
INSTANCE;
@Nonnull
@Override
public Completable convert(Publisher<?> source) {
return (Completable) getRequiredAdapter(Completable.class).fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link Observable}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava1ObservableConverter implements Converter<Publisher<?>, Observable<?>> {
INSTANCE;
@Nonnull
@Override
public Observable<?> convert(Publisher<?> source) {
return (Observable<?>) getRequiredAdapter(Observable.class).fromPublisher(Flux.from(source));
}
}
/**
* A {@link Converter} to convert a {@link Single} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1SingleToPublisherConverter implements Converter<Single<?>, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(Single<?> source) {
return Flux.defer(() -> getRequiredAdapter(Single.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link Single} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1SingleToMonoConverter implements Converter<Single<?>, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(Single<?> source) {
return Mono.defer(() -> Mono.from(getRequiredAdapter(Single.class).toPublisher(source)));
}
}
/**
* A {@link Converter} to convert a {@link Single} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1SingleToFluxConverter implements Converter<Single<?>, Flux<?>> {
INSTANCE;
@Nonnull
@Override
public Flux<?> convert(Single<?> source) {
return Flux.defer(() -> getRequiredAdapter(Single.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link Completable} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1CompletableToPublisherConverter implements Converter<Completable, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(Completable source) {
return Flux.defer(() -> getRequiredAdapter(Completable.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link Completable} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1CompletableToMonoConverter implements Converter<Completable, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(Completable source) {
return Mono.from(RxJava1CompletableToPublisherConverter.INSTANCE.convert(source));
}
}
/**
* A {@link Converter} to convert an {@link Observable} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1ObservableToPublisherConverter implements Converter<Observable<?>, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(Observable<?> source) {
return Flux.defer(() -> getRequiredAdapter(Observable.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1ObservableToMonoConverter implements Converter<Observable<?>, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(Observable<?> source) {
return Mono.defer(() -> Mono.from(getRequiredAdapter(Observable.class).toPublisher(source)));
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Flux}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1ObservableToFluxConverter implements Converter<Observable<?>, Flux<?>> {
INSTANCE;
@Nonnull
@Override
public Flux<?> convert(Observable<?> source) {
return Flux.defer(() -> getRequiredAdapter(Observable.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1ObservableToSingleConverter implements Converter<Observable<?>, Single<?>> {
INSTANCE;
@Nonnull
@Override
public Single<?> convert(Observable<?> source) {
return source.toSingle();
}
}
/**
* A {@link Converter} to convert a {@link Single} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava1SingleToObservableConverter implements Converter<Single<?>, Observable<?>> {
INSTANCE;
@Nonnull
@Override
public Observable<?> convert(Single<?> source) {
return source.toObservable();
}
}
// -------------------------------------------------------------------------
// RxJava 2 converters
// -------------------------------------------------------------------------
/**
* A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Single}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava2SingleConverter implements Converter<Publisher<?>, io.reactivex.Single<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Single<?> convert(Publisher<?> source) {
return (io.reactivex.Single<?>) getRequiredAdapter(io.reactivex.Single.class).fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Completable}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava2CompletableConverter implements Converter<Publisher<?>, io.reactivex.Completable> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Completable convert(Publisher<?> source) {
return (io.reactivex.Completable) getRequiredAdapter(io.reactivex.Completable.class).fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Observable}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava2ObservableConverter implements Converter<Publisher<?>, io.reactivex.Observable<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Observable<?> convert(Publisher<?> source) {
return (io.reactivex.Observable<?>) getRequiredAdapter(io.reactivex.Observable.class).fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Single} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2SingleToPublisherConverter implements Converter<io.reactivex.Single<?>, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(io.reactivex.Single<?> source) {
return getRequiredAdapter(io.reactivex.Single.class).toPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Single} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2SingleToMonoConverter implements Converter<io.reactivex.Single<?>, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(io.reactivex.Single<?> source) {
return Mono.from(getRequiredAdapter(io.reactivex.Single.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Single} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2SingleToFluxConverter implements Converter<io.reactivex.Single<?>, Flux<?>> {
INSTANCE;
@Nonnull
@Override
public Flux<?> convert(io.reactivex.Single<?> source) {
return Flux.from(getRequiredAdapter(io.reactivex.Single.class).toPublisher(source));
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Completable} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2CompletableToPublisherConverter implements Converter<io.reactivex.Completable, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(io.reactivex.Completable source) {
return getRequiredAdapter(io.reactivex.Completable.class).toPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Completable} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2CompletableToMonoConverter implements Converter<io.reactivex.Completable, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(io.reactivex.Completable source) {
return Mono.from(RxJava2CompletableToPublisherConverter.INSTANCE.convert(source));
}
}
/**
* A {@link Converter} to convert an {@link io.reactivex.Observable} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2ObservableToPublisherConverter implements Converter<io.reactivex.Observable<?>, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(io.reactivex.Observable<?> source) {
return source.toFlowable(BackpressureStrategy.BUFFER);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Observable} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2ObservableToMonoConverter implements Converter<io.reactivex.Observable<?>, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(io.reactivex.Observable<?> source) {
return Mono.from(source.toFlowable(BackpressureStrategy.BUFFER));
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Observable} to {@link Flux}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2ObservableToFluxConverter implements Converter<io.reactivex.Observable<?>, Flux<?>> {
INSTANCE;
@Nonnull
@Override
public Flux<?> convert(io.reactivex.Observable<?> source) {
return Flux.from(source.toFlowable(BackpressureStrategy.BUFFER));
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Flowable}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava2FlowableConverter implements Converter<Publisher<?>, io.reactivex.Flowable<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Flowable<?> convert(Publisher<?> source) {
return Flowable.fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Flowable} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2FlowableToPublisherConverter implements Converter<io.reactivex.Flowable<?>, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(io.reactivex.Flowable<?> source) {
return source;
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Flowable}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum PublisherToRxJava2MaybeConverter implements Converter<Publisher<?>, io.reactivex.Maybe<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Maybe<?> convert(Publisher<?> source) {
return (io.reactivex.Maybe<?>) getRequiredAdapter(Maybe.class).fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Maybe} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2MaybeToPublisherConverter implements Converter<io.reactivex.Maybe<?>, Publisher<?>> {
INSTANCE;
@Nonnull
@Override
public Publisher<?> convert(io.reactivex.Maybe<?> source) {
return source.toFlowable();
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Maybe} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2MaybeToMonoConverter implements Converter<io.reactivex.Maybe<?>, Mono<?>> {
INSTANCE;
@Nonnull
@Override
public Mono<?> convert(io.reactivex.Maybe<?> source) {
return Mono.from(source.toFlowable());
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Maybe} to {@link Flux}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2MaybeToFluxConverter implements Converter<io.reactivex.Maybe<?>, Flux<?>> {
INSTANCE;
@Nonnull
@Override
public Flux<?> convert(io.reactivex.Maybe<?> source) {
return Flux.from(source.toFlowable());
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2ObservableToSingleConverter
implements Converter<io.reactivex.Observable<?>, io.reactivex.Single<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Single<?> convert(io.reactivex.Observable<?> source) {
return source.singleOrError();
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Maybe}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2ObservableToMaybeConverter
implements Converter<io.reactivex.Observable<?>, io.reactivex.Maybe<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Maybe<?> convert(io.reactivex.Observable<?> source) {
return source.singleElement();
}
}
/**
* A {@link Converter} to convert a {@link Single} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
private enum RxJava2SingleToObservableConverter
implements Converter<io.reactivex.Single<?>, io.reactivex.Observable<?>> {
INSTANCE;
@Nonnull
@Override
public io.reactivex.Observable<?> convert(io.reactivex.Single<?> source) {
return source.toObservable();
}
}
/**
* A {@link ConverterFactory} that adapts between reactive types using {@link ReactiveAdapterRegistry}.
*/
@@ -1110,8 +458,8 @@ public class ReactiveWrapperConverters {
}
private boolean isSupported(TypeDescriptor typeDescriptor) {
return REACTIVE_ADAPTER_REGISTRY != null
&& REACTIVE_ADAPTER_REGISTRY.getAdapter(typeDescriptor.getType()) != null;
return RegistryHolder.REACTIVE_ADAPTER_REGISTRY != null
&& RegistryHolder.REACTIVE_ADAPTER_REGISTRY.getAdapter(typeDescriptor.getType()) != null;
}
@Override
@@ -1120,9 +468,9 @@ public class ReactiveWrapperConverters {
return source -> {
Publisher<?> publisher = source instanceof Publisher ? (Publisher<?>) source
: REACTIVE_ADAPTER_REGISTRY.getAdapter(Publisher.class, source).toPublisher(source);
: RegistryHolder.REACTIVE_ADAPTER_REGISTRY.getAdapter(Publisher.class, source).toPublisher(source);
ReactiveAdapter adapter = REACTIVE_ADAPTER_REGISTRY.getAdapter(targetType);
ReactiveAdapter adapter = RegistryHolder.REACTIVE_ADAPTER_REGISTRY.getAdapter(targetType);
return (T) adapter.fromPublisher(publisher);
};

93
src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java
View File

@@ -15,24 +15,17 @@
*/
package org.springframework.data.repository.util;
import kotlin.Unit;
import kotlinx.coroutines.flow.Flow;
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.Optional;
import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
import org.springframework.core.ReactiveAdapter;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.ReactiveTypeDescriptor;
import org.springframework.data.util.ProxyUtils;
import org.springframework.data.util.ReflectionUtils;
@@ -79,8 +72,6 @@ public class ReactiveWrappers {
&& ClassUtils.isPresent("kotlinx.coroutines.reactive.ReactiveFlowKt", ReactiveWrappers.class.getClassLoader())
&& ClassUtils.isPresent("kotlinx.coroutines.reactor.ReactorFlowKt", ReactiveWrappers.class.getClassLoader());
private static final Collection<ReactiveTypeDescriptor> REACTIVE_WRAPPERS;
/**
* Enumeration of supported reactive libraries.
*
@@ -90,47 +81,6 @@ public class ReactiveWrappers {
PROJECT_REACTOR, RXJAVA1, RXJAVA2, KOTLIN_COROUTINES;
}
static {
Collection<ReactiveTypeDescriptor> reactiveWrappers = new ArrayList<>(5);
if (RXJAVA1_PRESENT) {
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.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.add(ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty));
reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty));
reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty));
if (KOTLIN_COROUTINES_PRESENT) {
reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Flow.class, () -> {
return (Flow<Object>) (flowCollector, continuation) -> Unit.INSTANCE;
}));
}
}
REACTIVE_WRAPPERS = Collections.unmodifiableCollection(reactiveWrappers);
}
/**
* Returns {@literal true} if reactive support is available. More specifically, whether any of the libraries defined
* in {@link ReactiveLibrary} are on the class path.
@@ -172,7 +122,7 @@ public class ReactiveWrappers {
* @return {@literal true} if the {@code type} is a supported reactive wrapper type.
*/
public static boolean supports(Class<?> type) {
return isWrapper(ProxyUtils.getUserClass(type));
return isAvailable() && isWrapper(ProxyUtils.getUserClass(type));
}
/**
@@ -187,7 +137,7 @@ public class ReactiveWrappers {
return Arrays.stream(type.getMethods())//
.flatMap(ReflectionUtils::returnTypeAndParameters)//
.anyMatch(ReactiveWrappers::supports);
.anyMatch(ReactiveWrapperConverters::supports);
}
/**
@@ -238,38 +188,33 @@ public class ReactiveWrappers {
* Returns a collection of no-value wrapper types.
*
* @return a collection of no-value wrapper types.
* @deprecated not supported anymore.
*/
@Deprecated
public static Collection<Class<?>> getNoValueTypes() {
return REACTIVE_WRAPPERS.stream()//
.filter(ReactiveTypeDescriptor::isNoValue)//
.map(ReactiveTypeDescriptor::getReactiveType)//
.collect(Collectors.toList());
return Collections.emptyList();
}
/**
* Returns a collection of single-value wrapper types.
*
* @return a collection of single-value wrapper types.
* @deprecated not supported anymore.
*/
@Deprecated
public static Collection<Class<?>> getSingleValueTypes() {
return REACTIVE_WRAPPERS.stream()//
.filter(entry -> !entry.isMultiValue())//
.map(ReactiveTypeDescriptor::getReactiveType).collect(Collectors.toList());
return Collections.emptyList();
}
/**
* Returns a collection of multi-value wrapper types.
*
* @return a collection of multi-value wrapper types.
* @deprecated not supported anymore.
*/
@Deprecated
public static Collection<Class<?>> getMultiValueTypes() {
return REACTIVE_WRAPPERS.stream()//
.filter(ReactiveTypeDescriptor::isMultiValue)//
.map(ReactiveTypeDescriptor::getReactiveType)//
.collect(Collectors.toList());
return Collections.emptyList();
}
/**
@@ -295,8 +240,14 @@ public class ReactiveWrappers {
Assert.notNull(type, "Wrapper type must not be null!");
return REACTIVE_WRAPPERS.stream()//
.filter(it -> ClassUtils.isAssignable(it.getReactiveType(), type))//
.findFirst();
ReactiveAdapterRegistry adapterRegistry = ReactiveWrapperConverters.RegistryHolder.REACTIVE_ADAPTER_REGISTRY;
if (adapterRegistry == null) {
return Optional.empty();
}
ReactiveAdapter adapter = adapterRegistry.getAdapter(type);
return Optional.ofNullable(adapter == null ? null : adapter.getDescriptor());
}
}