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

DATACMNS-836 - Provide wrapper conversion, ReactiveWrappers and RxJava 2 type conversion.

Browse Source
This commit is contained in:
Mark Paluch
2016-07-30 18:40:52 -07:00
committed by Oliver Gierke
parent aa32c3d53d
commit 474c9981da
16 changed files with 1368 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/java/org/springframework/data/repository/config/RepositoryConfigurationExtensionSupport.java
View File

@@ -43,6 +43,7 @@ import org.springframework.util.StringUtils;
* {@link #getModulePrefix()}). Stubs out the post-processing methods as they might not be needed by default.
*
* @author Oliver Gierke
* @author Mark Paluch
*/
public abstract class RepositoryConfigurationExtensionSupport implements RepositoryConfigurationExtension {
@@ -293,7 +294,7 @@ public abstract class RepositoryConfigurationExtensionSupport implements Reposit
* @param loader must not be {@literal null}.
* @return the repository interface or {@literal null} if it can't be loaded.
*/
private Class<?> loadRepositoryInterface(RepositoryConfiguration<?> configuration, ResourceLoader loader) {
protected Class<?> loadRepositoryInterface(RepositoryConfiguration<?> configuration, ResourceLoader loader) {
String repositoryInterface = configuration.getRepositoryInterface();
ClassLoader classLoader = loader.getClassLoader();

29
src/main/java/org/springframework/data/repository/core/support/ReactiveRepositoryInformation.java
View File

@@ -33,6 +33,7 @@ import org.springframework.util.Assert;
* converted for invocation of implementation methods.
*
* @author Mark Paluch
* @since 2.0
*/
public class ReactiveRepositoryInformation extends DefaultRepositoryInformation {
@@ -75,7 +76,7 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
boolean wantsWrappers = wantsMethodUsingReactiveWrapperParameters(method);
if (wantsWrappers) {
Method candidate = getMethodCandidate(method, baseClass, new ExactWrapperMatch(method));
Method candidate = getMethodCandidate(method, baseClass, new AssignableWrapperMatch(method));
if (candidate != null) {
return candidate;
@@ -171,6 +172,11 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
&& !QueryExecutionConverters.supportsUnwrapping(parameterType);
}
/**
* {@link BiPredicate} to check whether a method parameter is a {@link #isNonunwrappingWrapper(Class)} and can be
* converted into a different wrapper. Usually {@link rx.Observable} to {@link org.reactivestreams.Publisher}
* conversion.
*/
static class WrapperConversionMatch implements BiPredicate<Class<?>, Integer> {
final Method declaredMethod;
@@ -187,7 +193,6 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
@Override
public boolean test(Class<?> candidateParameterType, Integer index) {
// TODO: should check for component type
if (isNonunwrappingWrapper(candidateParameterType) && isNonunwrappingWrapper(declaredParameterTypes[index])) {
if (conversionService.canConvert(declaredParameterTypes[index], candidateParameterType)) {
@@ -197,15 +202,19 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
return false;
}
}
static class ExactWrapperMatch implements BiPredicate<Class<?>, Integer> {
/**
* {@link BiPredicate} to check parameter assignability between a {@link #isNonunwrappingWrapper(Class)} parameter and
* a declared parameter. Usually {@link reactor.core.publisher.Flux} vs. {@link org.reactivestreams.Publisher}
* conversion.
*/
static class AssignableWrapperMatch implements BiPredicate<Class<?>, Integer> {
final Method declaredMethod;
final Class<?>[] declaredParameterTypes;
public ExactWrapperMatch(Method declaredMethod) {
public AssignableWrapperMatch(Method declaredMethod) {
this.declaredMethod = declaredMethod;
this.declaredParameterTypes = declaredMethod.getParameterTypes();
@@ -214,7 +223,6 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
@Override
public boolean test(Class<?> candidateParameterType, Integer index) {
// TODO: should check for component type
if (isNonunwrappingWrapper(candidateParameterType) && isNonunwrappingWrapper(declaredParameterTypes[index])) {
if (declaredParameterTypes[index].isAssignableFrom(candidateParameterType)) {
@@ -224,9 +232,14 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
return false;
}
}
/**
* {@link BiPredicate} to check parameter assignability between a parameters in which the declared parameter may be
* wrapped but supports unwrapping. Usually types like {@link java.util.Optional} or {@link java.util.stream.Stream}.
*
* @see QueryExecutionConverters
*/
static class MatchParameterOrComponentType implements BiPredicate<Class<?>, Integer> {
final Method declaredMethod;
@@ -253,7 +266,5 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation
return true;
}
}
}

267
src/main/java/org/springframework/data/repository/query/ReactiveWrapperConverters.java
View File

@@ -1,267 +0,0 @@
/*
* Copyright 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.data.repository.query;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.function.Predicate;
import org.reactivestreams.Publisher;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.data.repository.util.QueryExecutionConverters;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.Single;
/**
* Conversion support for reactive wrapper types.
*
* @author Mark Paluch
* @since 2.0
*/
public abstract class ReactiveWrapperConverters {
private static final boolean PROJECT_REACTOR_PRESENT = ClassUtils.isPresent("reactor.core.converter.DependencyUtils",
QueryExecutionConverters.class.getClassLoader());
private static final boolean RXJAVA_SINGLE_PRESENT = ClassUtils.isPresent("rx.Single",
QueryExecutionConverters.class.getClassLoader());
private static final boolean RXJAVA_OBSERVABLE_PRESENT = ClassUtils.isPresent("rx.Observable",
QueryExecutionConverters.class.getClassLoader());
private static final List<AbstractReactiveWrapper<?>> REACTIVE_WRAPPERS = new ArrayList<>();
private static final GenericConversionService GENERIC_CONVERSION_SERVICE = new GenericConversionService();
static {
if (PROJECT_REACTOR_PRESENT) {
REACTIVE_WRAPPERS.add(FluxWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(MonoWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(PublisherWrapper.INSTANCE);
}
if (RXJAVA_SINGLE_PRESENT) {
REACTIVE_WRAPPERS.add(SingleWrapper.INSTANCE);
}
if (RXJAVA_OBSERVABLE_PRESENT) {
REACTIVE_WRAPPERS.add(ObservableWrapper.INSTANCE);
}
QueryExecutionConverters.registerConvertersIn(GENERIC_CONVERSION_SERVICE);
}
private ReactiveWrapperConverters() {
}
/**
* Returns whether the given type is a supported wrapper type.
*
* @param type must not be {@literal null}.
* @return
*/
public static boolean supports(Class<?> type) {
return assignableStream(type).isPresent();
}
/**
* Returns whether the type is a single-like wrapper.
*
* @param type must not be {@literal null}.
* @return
* @see Single
* @see Mono
*/
public static boolean isSingleLike(Class<?> type) {
return assignableStream(type).map(wrapper -> wrapper.getMultiplicity() == Multiplicity.ONE).orElse(false);
}
/**
* Returns whether the type is a collection/multi-element-like wrapper.
*
* @param type must not be {@literal null}.
* @return
* @see Observable
* @see Flux
* @see Publisher
*/
public static boolean isCollectionLike(Class<?> type) {
return assignableStream(type).map(wrapper -> wrapper.getMultiplicity() == Multiplicity.MANY).orElse(false);
}
/**
* Casts or converts the given wrapper type into a different wrapper type.
*
* @param stream the stream, must not be {@literal null}.
* @param expectedWrapperType must not be {@literal null}.
* @return
*/
public static <T> T toWrapper(Object stream, Class<? extends T> expectedWrapperType) {
Assert.notNull(stream, "Stream must not be null!");
Assert.notNull(expectedWrapperType, "Converter must not be null!");
if (expectedWrapperType.isAssignableFrom(stream.getClass())) {
return (T) stream;
}
return GENERIC_CONVERSION_SERVICE.convert(stream, expectedWrapperType);
}
/**
* Maps elements of a reactive element stream to other elements.
*
* @param stream must not be {@literal null}.
* @param converter must not be {@literal null}.
* @return
*/
public static <T> T map(Object stream, Converter<Object, Object> converter) {
Assert.notNull(stream, "Stream must not be null!");
Assert.notNull(converter, "Converter must not be null!");
for (AbstractReactiveWrapper<?> reactiveWrapper : REACTIVE_WRAPPERS) {
if (ClassUtils.isAssignable(reactiveWrapper.getWrapperClass(), stream.getClass())) {
return (T) reactiveWrapper.map(stream, converter);
}
}
throw new IllegalStateException(String.format("Cannot apply converter to %s", stream));
}
private static Optional<AbstractReactiveWrapper<?>> assignableStream(Class<?> type) {
Assert.notNull(type, "Type must not be null!");
return findWrapper(wrapper -> ClassUtils.isAssignable(wrapper.getWrapperClass(), type));
}
private static Optional<AbstractReactiveWrapper<?>> findWrapper(
Predicate<? super AbstractReactiveWrapper<?>> predicate) {
return REACTIVE_WRAPPERS.stream().filter(predicate).findFirst();
}
private abstract static class AbstractReactiveWrapper<T> {
private final Class<? super T> wrapperClass;
private final Multiplicity multiplicity;
public AbstractReactiveWrapper(Class<? super T> wrapperClass, Multiplicity multiplicity) {
this.wrapperClass = wrapperClass;
this.multiplicity = multiplicity;
}
public Class<? super T> getWrapperClass() {
return wrapperClass;
}
public Multiplicity getMultiplicity() {
return multiplicity;
}
public abstract Object map(Object wrapper, Converter<Object, Object> converter);
}
private static class MonoWrapper extends AbstractReactiveWrapper<Mono<?>> {
static final MonoWrapper INSTANCE = new MonoWrapper();
private MonoWrapper() {
super(Mono.class, Multiplicity.ONE);
}
public Mono<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Mono<?>) wrapper).map(converter::convert);
}
}
private static class FluxWrapper extends AbstractReactiveWrapper<Flux<?>> {
static final FluxWrapper INSTANCE = new FluxWrapper();
private FluxWrapper() {
super(Flux.class, Multiplicity.MANY);
}
public Flux<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Flux<?>) wrapper).map(converter::convert);
}
}
private static class PublisherWrapper extends AbstractReactiveWrapper<Publisher<?>> {
static final PublisherWrapper INSTANCE = new PublisherWrapper();
public PublisherWrapper() {
super(Publisher.class, Multiplicity.MANY);
}
@Override
public Publisher<?> map(Object wrapper, Converter<Object, Object> converter) {
if (wrapper instanceof Mono) {
return MonoWrapper.INSTANCE.map((Mono<?>) wrapper, converter);
}
if (wrapper instanceof Flux) {
return FluxWrapper.INSTANCE.map((Flux<?>) wrapper, converter);
}
return FluxWrapper.INSTANCE.map(Flux.from((Publisher<?>) wrapper), converter);
}
}
private static class SingleWrapper extends AbstractReactiveWrapper<Single<?>> {
static final SingleWrapper INSTANCE = new SingleWrapper();
private SingleWrapper() {
super(Single.class, Multiplicity.ONE);
}
@Override
public Single<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Single<?>) wrapper).map(converter::convert);
}
}
private static class ObservableWrapper extends AbstractReactiveWrapper<Observable<?>> {
static final ObservableWrapper INSTANCE = new ObservableWrapper();
private ObservableWrapper() {
super(Observable.class, Multiplicity.MANY);
}
@Override
public Observable<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Observable<?>) wrapper).map(converter::convert);
}
}
private enum Multiplicity {
ONE, MANY,
}
}

21
src/main/java/org/springframework/data/repository/query/ResultProcessor.java
View File

@@ -32,6 +32,7 @@ import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.data.domain.Slice;
import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.repository.util.ReactiveWrapperConverters;
import org.springframework.data.util.ReflectionUtils;
import org.springframework.util.Assert;
@@ -54,7 +55,7 @@ public class ResultProcessor {
/**
* Creates a new {@link ResultProcessor} from the given {@link QueryMethod} and {@link ProjectionFactory}.
*
*
* @param method must not be {@literal null}.
* @param factory must not be {@literal null}.
*/
@@ -64,7 +65,7 @@ public class ResultProcessor {
/**
* Creates a new {@link ResultProcessor} for the given {@link QueryMethod}, {@link ProjectionFactory} and type.
*
*
* @param method must not be {@literal null}.
* @param factory must not be {@literal null}.
* @param type must not be {@literal null}.
@@ -83,7 +84,7 @@ public class ResultProcessor {
/**
* Returns a new {@link ResultProcessor} with a new projection type obtained from the given {@link ParameterAccessor}.
*
*
* @param accessor can be {@literal null}.
* @return
*/
@@ -100,7 +101,7 @@ public class ResultProcessor {
/**
* Returns the {@link ReturnedType}.
*
*
* @return
*/
public ReturnedType getReturnedType() {
@@ -109,7 +110,7 @@ public class ResultProcessor {
/**
* Post-processes the given query result.
*
*
* @param source can be {@literal null}.
* @return
*/
@@ -120,7 +121,7 @@ public class ResultProcessor {
/**
* Post-processes the given query result using the given preparing {@link Converter} to potentially prepare collection
* elements.
*
*
* @param source can be {@literal null}.
* @param preparingConverter must not be {@literal null}.
* @return
@@ -189,7 +190,7 @@ public class ResultProcessor {
/**
* Returns a new {@link ChainingConverter} that hands the elements resulting from the current conversion to the
* given {@link Converter}.
*
*
* @param converter must not be {@literal null}.
* @return
*/
@@ -208,7 +209,7 @@ public class ResultProcessor {
});
}
/*
/*
* (non-Javadoc)
* @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
*/
@@ -228,7 +229,7 @@ public class ResultProcessor {
INSTANCE;
/*
/*
* (non-Javadoc)
* @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
*/
@@ -245,7 +246,7 @@ public class ResultProcessor {
private final @NonNull ProjectionFactory factory;
private final ConversionService conversionService = new DefaultConversionService();
/*
/*
* (non-Javadoc)
* @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
*/

2
src/main/java/org/springframework/data/repository/reactive/ReactiveCrudRepository.java
View File

@@ -29,9 +29,9 @@ import reactor.core.publisher.Mono;
* and uses Project Reactor types which are built on top of Reactive Streams.
*
* @author Mark Paluch
* @since 2.0
* @see Mono
* @see Flux
* @see ReactiveStreamsCrudRepository
*/
@NoRepositoryBean
public interface ReactiveCrudRepository<T, ID extends Serializable> extends Repository<T, ID> {

1
src/main/java/org/springframework/data/repository/reactive/ReactivePagingAndSortingRepository.java
View File

@@ -31,6 +31,7 @@ import reactor.core.publisher.Mono;
* and sorting abstraction.
*
* @author Mark Paluch
* @since 2.0
* @see Sort
* @see Pageable
* @see Mono

3
src/main/java/org/springframework/data/repository/reactive/RxJavaCrudRepository.java
View File

@@ -26,9 +26,10 @@ import rx.Single;
/**
* Interface for generic CRUD operations on a repository for a specific type. This repository follows reactive paradigms
* and uses RxJava types.
* and uses RxJava 1 types.
*
* @author Mark Paluch
* @since 2.0
* @see Single
* @see Observable
*/

1
src/main/java/org/springframework/data/repository/reactive/RxJavaPagingAndSortingRepository.java
View File

@@ -31,6 +31,7 @@ import rx.Single;
* abstraction.
*
* @author Mark Paluch
* @since 2.0
* @see Sort
* @see Pageable
* @see Single

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

@@ -15,10 +15,6 @@
*/
package org.springframework.data.repository.util;
import scala.Function0;
import scala.Option;
import scala.runtime.AbstractFunction0;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
@@ -26,6 +22,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import org.reactivestreams.Publisher;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.core.convert.converter.Converter;
@@ -38,13 +35,17 @@ import org.springframework.util.concurrent.ListenableFuture;
import com.google.common.base.Optional;
import reactor.core.converter.DependencyUtils;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Completable;
import rx.Observable;
import rx.Single;
import scala.Function0;
import scala.Option;
import scala.runtime.AbstractFunction0;
/**
* Converters to potentially wrap the execution of a repository method into a variety of wrapper types potentially being
@@ -56,17 +57,13 @@ import scala.Option;
* <li>{@code java.util.concurrent.Future}</li>
* <li>{@code java.util.concurrent.CompletableFuture}</li>
* <li>{@code org.springframework.util.concurrent.ListenableFuture<}</li>
* <li>{@code rx.Single}</li>
* <li>{@code rx.Observable}</li>
* <li>{@code rx.Completable}</li>
* <li>{@code reactor.core.publisher.Mono}</li>
* <li>{@code reactor.core.publisher.Flux}</li>
* <li>{@code org.reactivestreams.Publisher}</li>
* <li>Reactive wrappers supported by {@link ReactiveWrappers}</li>
* </ul>
*
* @author Oliver Gierke
* @author Mark Paluch
* @since 1.8
* @see ReactiveWrappers
*/
public abstract class QueryExecutionConverters {
@@ -74,6 +71,9 @@ public abstract class QueryExecutionConverters {
"org.springframework.core.annotation.AnnotationConfigurationException",
QueryExecutionConverters.class.getClassLoader());
private static final boolean ASYNC_RESULT_PRESENT = ClassUtils.isPresent(
"org.springframework.scheduling.annotation.AsyncResult", QueryExecutionConverters.class.getClassLoader());
private static final boolean GUAVA_PRESENT = ClassUtils.isPresent("com.google.common.base.Optional",
QueryExecutionConverters.class.getClassLoader());
private static final boolean JDK_8_PRESENT = ClassUtils.isPresent("java.util.Optional",
@@ -81,18 +81,10 @@ public abstract class QueryExecutionConverters {
private static final boolean SCALA_PRESENT = ClassUtils.isPresent("scala.Option",
QueryExecutionConverters.class.getClassLoader());
private static final boolean PROJECT_REACTOR_PRESENT = ClassUtils.isPresent("reactor.core.converter.DependencyUtils",
QueryExecutionConverters.class.getClassLoader());
private static final boolean RXJAVA_SINGLE_PRESENT = ClassUtils.isPresent("rx.Single",
QueryExecutionConverters.class.getClassLoader());
private static final boolean RXJAVA_OBSERVABLE_PRESENT = ClassUtils.isPresent("rx.Observable",
QueryExecutionConverters.class.getClassLoader());
private static final boolean RXJAVA_COMPLETABLE_PRESENT = ClassUtils.isPresent("rx.Completable",
QueryExecutionConverters.class.getClassLoader());
private static final Set<Class<?>> WRAPPER_TYPES = new HashSet<Class<?>>();
private static final Set<Class<?>> UNWRAPPER_TYPES = new HashSet<Class<?>>();
private static final Set<Converter<Object, Object>> UNWRAPPERS = new HashSet<Converter<Object, Object>>();
private static final ReactiveAdapterRegistry REACTIVE_ADAPTER_REGISTRY = new ReactiveAdapterRegistry();
static {
@@ -124,23 +116,9 @@ public abstract class QueryExecutionConverters {
UNWRAPPERS.add(ScalOptionUnwrapper.INSTANCE);
}
if (PROJECT_REACTOR_PRESENT) {
WRAPPER_TYPES.add(Publisher.class);
WRAPPER_TYPES.add(Mono.class);
WRAPPER_TYPES.add(Flux.class);
}
if (RXJAVA_SINGLE_PRESENT) {
WRAPPER_TYPES.add(Single.class);
}
if (RXJAVA_COMPLETABLE_PRESENT) {
WRAPPER_TYPES.add(Completable.class);
}
if (RXJAVA_OBSERVABLE_PRESENT) {
WRAPPER_TYPES.add(Observable.class);
}
WRAPPER_TYPES.addAll(ReactiveWrappers.getNoValueTypes());
WRAPPER_TYPES.addAll(ReactiveWrappers.getSingleValueTypes());
WRAPPER_TYPES.addAll(ReactiveWrappers.getMultiValueTypes());
}
private QueryExecutionConverters() {}
@@ -205,37 +183,68 @@ public abstract class QueryExecutionConverters {
conversionService.addConverter(new NullableWrapperToScalaOptionConverter(conversionService));
}
conversionService.addConverter(new NullableWrapperToFutureConverter(conversionService));
if (PROJECT_REACTOR_PRESENT) {
if (RXJAVA_COMPLETABLE_PRESENT) {
conversionService.addConverter(PublisherToCompletableConverter.INSTANCE);
conversionService.addConverter(CompletableToPublisherConverter.INSTANCE);
conversionService.addConverter(CompletableToMonoConverter.INSTANCE);
}
if (RXJAVA_SINGLE_PRESENT) {
conversionService.addConverter(PublisherToSingleConverter.INSTANCE);
conversionService.addConverter(SingleToPublisherConverter.INSTANCE);
conversionService.addConverter(SingleToMonoConverter.INSTANCE);
conversionService.addConverter(SingleToFluxConverter.INSTANCE);
}
if (RXJAVA_OBSERVABLE_PRESENT) {
conversionService.addConverter(PublisherToObservableConverter.INSTANCE);
conversionService.addConverter(ObservableToPublisherConverter.INSTANCE);
conversionService.addConverter(ObservableToMonoConverter.INSTANCE);
conversionService.addConverter(ObservableToFluxConverter.INSTANCE);
}
conversionService.addConverter(PublisherToMonoConverter.INSTANCE);
conversionService.addConverter(PublisherToFluxConverter.INSTANCE);
if (ASYNC_RESULT_PRESENT) {
conversionService.addConverter(new NullableWrapperToFutureConverter(conversionService));
}
if (RXJAVA_SINGLE_PRESENT && RXJAVA_OBSERVABLE_PRESENT) {
conversionService.addConverter(SingleToObservableConverter.INSTANCE);
conversionService.addConverter(ObservableToSingleConverter.INSTANCE);
if (ReactiveWrappers.isAvailable()) {
if (ReactiveWrappers.RXJAVA1_PRESENT) {
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.RXJAVA2_PRESENT) {
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);
}
if (ReactiveWrappers.PROJECT_REACTOR_PRESENT) {
conversionService.addConverter(PublisherToMonoConverter.INSTANCE);
conversionService.addConverter(PublisherToFluxConverter.INSTANCE);
}
if (ReactiveWrappers.RXJAVA1_PRESENT) {
conversionService.addConverter(RxJava1SingleToObservableConverter.INSTANCE);
conversionService.addConverter(RxJava1ObservableToSingleConverter.INSTANCE);
}
if (ReactiveWrappers.RXJAVA2_PRESENT) {
conversionService.addConverter(RxJava2SingleToObservableConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToSingleConverter.INSTANCE);
}
}
}
@@ -585,13 +594,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum PublisherToSingleConverter implements Converter<Publisher<?>, Single<?>> {
public enum PublisherToRxJava1SingleConverter implements Converter<Publisher<?>, Single<?>> {
INSTANCE;
@Override
public Single<?> convert(Publisher<?> source) {
return DependencyUtils.convertFromPublisher(source, Single.class);
return (Single<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Single.class).fromPublisher(Mono.from(source));
}
}
@@ -601,13 +610,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum PublisherToCompletableConverter implements Converter<Publisher<?>, Completable> {
public enum PublisherToRxJava1CompletableConverter implements Converter<Publisher<?>, Completable> {
INSTANCE;
@Override
public Completable convert(Publisher<?> source) {
return DependencyUtils.convertFromPublisher(source, Completable.class);
return (Completable) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Completable.class).fromPublisher(source);
}
}
@@ -617,13 +626,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum PublisherToObservableConverter implements Converter<Publisher<?>, Observable<?>> {
public enum PublisherToRxJava1ObservableConverter implements Converter<Publisher<?>, Observable<?>> {
INSTANCE;
@Override
public Observable<?> convert(Publisher<?> source) {
return DependencyUtils.convertFromPublisher(source, Observable.class);
return (Observable<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Observable.class).fromPublisher(Flux.from(source));
}
}
@@ -633,13 +642,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum SingleToPublisherConverter implements Converter<Single<?>, Publisher<?>> {
public enum RxJava1SingleToPublisherConverter implements Converter<Single<?>, Publisher<?>> {
INSTANCE;
@Override
public Publisher<?> convert(Single<?> source) {
return DependencyUtils.convertToPublisher(source);
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toPublisher(source);
}
}
@@ -649,13 +658,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum SingleToMonoConverter implements Converter<Single<?>, Mono<?>> {
public enum RxJava1SingleToMonoConverter implements Converter<Single<?>, Mono<?>> {
INSTANCE;
@Override
public Mono<?> convert(Single<?> source) {
return PublisherToMonoConverter.INSTANCE.convert(DependencyUtils.convertToPublisher(source));
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toMono(source);
}
}
@@ -665,13 +674,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum SingleToFluxConverter implements Converter<Single<?>, Flux<?>> {
public enum RxJava1SingleToFluxConverter implements Converter<Single<?>, Flux<?>> {
INSTANCE;
@Override
public Flux<?> convert(Single<?> source) {
return PublisherToFluxConverter.INSTANCE.convert(DependencyUtils.convertToPublisher(source));
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toFlux(source);
}
}
@@ -681,13 +690,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum CompletableToPublisherConverter implements Converter<Completable, Publisher<?>> {
public enum RxJava1CompletableToPublisherConverter implements Converter<Completable, Publisher<?>> {
INSTANCE;
@Override
public Publisher<?> convert(Completable source) {
return DependencyUtils.convertToPublisher(source);
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Completable.class).toFlux(source);
}
}
@@ -697,13 +706,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum CompletableToMonoConverter implements Converter<Completable, Mono<?>> {
public enum RxJava1CompletableToMonoConverter implements Converter<Completable, Mono<?>> {
INSTANCE;
@Override
public Mono<?> convert(Completable source) {
return Mono.from(CompletableToPublisherConverter.INSTANCE.convert(source));
return Mono.from(RxJava1CompletableToPublisherConverter.INSTANCE.convert(source));
}
}
@@ -713,13 +722,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum ObservableToPublisherConverter implements Converter<Observable<?>, Publisher<?>> {
public enum RxJava1ObservableToPublisherConverter implements Converter<Observable<?>, Publisher<?>> {
INSTANCE;
@Override
public Publisher<?> convert(Observable<?> source) {
return DependencyUtils.convertToPublisher(source);
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source);
}
}
@@ -729,13 +738,13 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum ObservableToMonoConverter implements Converter<Observable<?>, Mono<?>> {
public enum RxJava1ObservableToMonoConverter implements Converter<Observable<?>, Mono<?>> {
INSTANCE;
@Override
public Mono<?> convert(Observable<?> source) {
return PublisherToMonoConverter.INSTANCE.convert(DependencyUtils.convertToPublisher(source));
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toMono(source);
}
}
@@ -745,13 +754,288 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum ObservableToFluxConverter implements Converter<Observable<?>, Flux<?>> {
public enum RxJava1ObservableToFluxConverter implements Converter<Observable<?>, Flux<?>> {
INSTANCE;
@Override
public Flux<?> convert(Observable<?> source) {
return PublisherToFluxConverter.INSTANCE.convert(DependencyUtils.convertToPublisher(source));
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source);
}
}
/**
* A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Single}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum PublisherToRxJava2SingleConverter implements Converter<Publisher<?>, io.reactivex.Single<?>> {
INSTANCE;
@Override
public io.reactivex.Single<?> convert(Publisher<?> source) {
return (io.reactivex.Single<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(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
*/
public enum PublisherToRxJava2CompletableConverter implements Converter<Publisher<?>, io.reactivex.Completable> {
INSTANCE;
@Override
public io.reactivex.Completable convert(Publisher<?> source) {
return (io.reactivex.Completable) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(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
*/
public enum PublisherToRxJava2ObservableConverter implements Converter<Publisher<?>, io.reactivex.Observable<?>> {
INSTANCE;
@Override
public io.reactivex.Observable<?> convert(Publisher<?> source) {
return (io.reactivex.Observable<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(io.reactivex.Single.class)
.fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Single} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2SingleToPublisherConverter implements Converter<io.reactivex.Single<?>, Publisher<?>> {
INSTANCE;
@Override
public Publisher<?> convert(io.reactivex.Single<?> source) {
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Single.class).toMono(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Single} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2SingleToMonoConverter implements Converter<io.reactivex.Single<?>, Mono<?>> {
INSTANCE;
@Override
public Mono<?> convert(io.reactivex.Single<?> source) {
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Single.class).toMono(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Single} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2SingleToFluxConverter implements Converter<io.reactivex.Single<?>, Flux<?>> {
INSTANCE;
@Override
public Flux<?> convert(io.reactivex.Single<?> source) {
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Single.class).toFlux(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Completable} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2CompletableToPublisherConverter implements Converter<io.reactivex.Completable, Publisher<?>> {
INSTANCE;
@Override
public Publisher<?> convert(io.reactivex.Completable source) {
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(io.reactivex.Completable.class).toFlux(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Completable} to {@link Mono}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2CompletableToMonoConverter implements Converter<io.reactivex.Completable, Mono<?>> {
INSTANCE;
@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
*/
public enum RxJava2ObservableToPublisherConverter implements Converter<io.reactivex.Observable<?>, Publisher<?>> {
INSTANCE;
@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
*/
public enum RxJava2ObservableToMonoConverter implements Converter<io.reactivex.Observable<?>, Mono<?>> {
INSTANCE;
@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
*/
public enum RxJava2ObservableToFluxConverter implements Converter<io.reactivex.Observable<?>, Flux<?>> {
INSTANCE;
@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
*/
public enum PublisherToRxJava2FlowableConverter implements Converter<Publisher<?>, io.reactivex.Flowable<?>> {
INSTANCE;
@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
*/
public enum RxJava2FlowableToPublisherConverter implements Converter<io.reactivex.Flowable<?>, Publisher<?>> {
INSTANCE;
@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
*/
public enum PublisherToRxJava2MaybeConverter implements Converter<Publisher<?>, io.reactivex.Maybe<?>> {
INSTANCE;
@Override
public io.reactivex.Maybe<?> convert(Publisher<?> source) {
return (io.reactivex.Maybe<?>) REACTIVE_ADAPTER_REGISTRY.getAdapterTo(Maybe.class).fromPublisher(source);
}
}
/**
* A {@link Converter} to convert a {@link io.reactivex.Maybe} to {@link Publisher}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2MaybeToPublisherConverter implements Converter<io.reactivex.Maybe<?>, Publisher<?>> {
INSTANCE;
@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
*/
public enum RxJava2MaybeToMonoConverter implements Converter<io.reactivex.Maybe<?>, Mono<?>> {
INSTANCE;
@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
*/
public enum RxJava2MaybeToFluxConverter implements Converter<io.reactivex.Maybe<?>, Flux<?>> {
INSTANCE;
@Override
public Flux<?> convert(io.reactivex.Maybe<?> source) {
return Flux.from(source.toFlowable());
}
}
@@ -761,7 +1045,7 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum ObservableToSingleConverter implements Converter<Observable<?>, Single<?>> {
public enum RxJava1ObservableToSingleConverter implements Converter<Observable<?>, Single<?>> {
INSTANCE;
@@ -777,7 +1061,7 @@ public abstract class QueryExecutionConverters {
* @author Mark Paluch
* @author 2.0
*/
public enum SingleToObservableConverter implements Converter<Single<?>, Observable<?>> {
public enum RxJava1SingleToObservableConverter implements Converter<Single<?>, Observable<?>> {
INSTANCE;
@@ -786,4 +1070,38 @@ public abstract class QueryExecutionConverters {
return source.toObservable();
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2ObservableToSingleConverter
implements Converter<io.reactivex.Observable<?>, io.reactivex.Single<?>> {
INSTANCE;
@Override
public io.reactivex.Single<?> convert(io.reactivex.Observable<?> source) {
return source.singleElement().toSingle();
}
}
/**
* A {@link Converter} to convert a {@link Single} to {@link Single}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2SingleToObservableConverter
implements Converter<io.reactivex.Single<?>, io.reactivex.Observable<?>> {
INSTANCE;
@Override
public io.reactivex.Observable<?> convert(io.reactivex.Single<?> source) {
return source.toObservable();
}
}
}

285
src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java Normal file
View File

@@ -0,0 +1,285 @@
/*
* Copyright 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.data.repository.util;
import java.util.ArrayList;
import java.util.List;
import org.reactivestreams.Publisher;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import io.reactivex.Flowable;
import lombok.experimental.UtilityClass;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.Single;
/**
* Conversion support for reactive wrapper types. This class is a logical extension to {@link QueryExecutionConverters}.
* <p>
* This class discovers reactive wrapper availability and their conversion support based on the class path. Reactive
* wrapper types might be supported/on the class path but conversion may require additional dependencies.
*
* @author Mark Paluch
* @since 2.0
* @see ReactiveWrappers
* @see ReactiveAdapterRegistry
*/
@UtilityClass
public class ReactiveWrapperConverters {
private static final List<AbstractReactiveWrapper<?>> REACTIVE_WRAPPERS = new ArrayList<>();
private static final GenericConversionService GENERIC_CONVERSION_SERVICE = new GenericConversionService();
private static final ReactiveAdapterRegistry REACTIVE_ADAPTER_REGISTRY = new ReactiveAdapterRegistry();
static {
if (ReactiveWrappers.RXJAVA1_PRESENT) {
REACTIVE_WRAPPERS.add(RxJava1SingleWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(RxJava1ObservableWrapper.INSTANCE);
}
if (ReactiveWrappers.RXJAVA2_PRESENT) {
REACTIVE_WRAPPERS.add(RxJava2SingleWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(RxJava2MaybeWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(RxJava2ObservableWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(RxJava2FlowableWrapper.INSTANCE);
}
if (ReactiveWrappers.PROJECT_REACTOR_PRESENT) {
REACTIVE_WRAPPERS.add(FluxWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(MonoWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(PublisherWrapper.INSTANCE);
}
QueryExecutionConverters.registerConvertersIn(GENERIC_CONVERSION_SERVICE);
}
/**
* Returns whether the given type is supported for wrapper type conversion.
* <p>
* NOTE: A reactive wrapper type might be supported in general by {@link ReactiveWrappers#supports(Class)} but not
* necessarily for conversion using this method.
* </p>
*
* @param type must not be {@literal null}.
* @return {@literal true} if the {@code type} is a supported reactive wrapper type.
*/
public static boolean supports(Class<?> type) {
return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(type) != null;
}
/**
* Casts or converts the given wrapper type into a different wrapper type.
*
* @param stream the stream, must not be {@literal null}.
* @param expectedWrapperType must not be {@literal null}.
* @return
*/
@SuppressWarnings("unchecked")
public static <T> T toWrapper(Object stream, Class<? extends T> expectedWrapperType) {
Assert.notNull(stream, "Stream must not be null!");
Assert.notNull(expectedWrapperType, "Converter must not be null!");
if (expectedWrapperType.isAssignableFrom(stream.getClass())) {
return (T) stream;
}
return GENERIC_CONVERSION_SERVICE.convert(stream, expectedWrapperType);
}
/**
* Maps elements of a reactive element stream to other elements.
*
* @param stream must not be {@literal null}.
* @param converter must not be {@literal null}.
* @return
*/
@SuppressWarnings("unchecked")
public static <T> T map(Object stream, Converter<Object, Object> converter) {
Assert.notNull(stream, "Stream must not be null!");
Assert.notNull(converter, "Converter must not be null!");
for (AbstractReactiveWrapper<?> reactiveWrapper : REACTIVE_WRAPPERS) {
if (ClassUtils.isAssignable(reactiveWrapper.getWrapperClass(), stream.getClass())) {
return (T) reactiveWrapper.map(stream, converter);
}
}
throw new IllegalStateException(String.format("Cannot apply converter to %s", stream));
}
private interface AbstractReactiveWrapper<T> {
Class<? super T> getWrapperClass();
Object map(Object wrapper, Converter<Object, Object> converter);
}
private enum MonoWrapper implements AbstractReactiveWrapper<Mono<?>> {
INSTANCE;
@Override
public Class<? super Mono<?>> getWrapperClass() {
return Mono.class;
}
@Override
public Mono<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Mono<?>) wrapper).map(converter::convert);
}
}
private enum FluxWrapper implements AbstractReactiveWrapper<Flux<?>> {
INSTANCE;
@Override
public Class<? super Flux<?>> getWrapperClass() {
return Flux.class;
}
public Flux<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Flux<?>) wrapper).map(converter::convert);
}
}
private enum PublisherWrapper implements AbstractReactiveWrapper<Publisher<?>> {
INSTANCE;
@Override
public Class<? super Publisher<?>> getWrapperClass() {
return Publisher.class;
}
@Override
public Publisher<?> map(Object wrapper, Converter<Object, Object> converter) {
if (wrapper instanceof Mono) {
return MonoWrapper.INSTANCE.map(wrapper, converter);
}
if (wrapper instanceof Flux) {
return FluxWrapper.INSTANCE.map(wrapper, converter);
}
return FluxWrapper.INSTANCE.map(Flux.from((Publisher<?>) wrapper), converter);
}
}
private enum RxJava1SingleWrapper implements AbstractReactiveWrapper<Single<?>> {
INSTANCE;
@Override
public Class<? super Single<?>> getWrapperClass() {
return Single.class;
}
@Override
public Single<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Single<?>) wrapper).map(converter::convert);
}
}
private enum RxJava1ObservableWrapper implements AbstractReactiveWrapper<Observable<?>> {
INSTANCE;
@Override
public Class<? super Observable<?>> getWrapperClass() {
return Observable.class;
}
@Override
public Observable<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((Observable<?>) wrapper).map(converter::convert);
}
}
private enum RxJava2SingleWrapper implements AbstractReactiveWrapper<io.reactivex.Single<?>> {
INSTANCE;
@Override
public Class<? super io.reactivex.Single<?>> getWrapperClass() {
return io.reactivex.Single.class;
}
@Override
public io.reactivex.Single<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((io.reactivex.Single<?>) wrapper).map(converter::convert);
}
}
private enum RxJava2MaybeWrapper implements AbstractReactiveWrapper<io.reactivex.Maybe<?>> {
INSTANCE;
@Override
public Class<? super io.reactivex.Maybe<?>> getWrapperClass() {
return io.reactivex.Maybe.class;
}
@Override
public io.reactivex.Maybe<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((io.reactivex.Maybe<?>) wrapper).map(converter::convert);
}
}
private enum RxJava2ObservableWrapper implements AbstractReactiveWrapper<io.reactivex.Observable<?>> {
INSTANCE;
@Override
public Class<? super io.reactivex.Observable<?>> getWrapperClass() {
return io.reactivex.Observable.class;
}
@Override
public io.reactivex.Observable<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((io.reactivex.Observable<?>) wrapper).map(converter::convert);
}
}
private enum RxJava2FlowableWrapper implements AbstractReactiveWrapper<io.reactivex.Flowable<?>> {
INSTANCE;
@Override
public Class<? super Flowable<?>> getWrapperClass() {
return io.reactivex.Flowable.class;
}
@Override
public io.reactivex.Flowable<?> map(Object wrapper, Converter<Object, Object> converter) {
return ((io.reactivex.Flowable<?>) wrapper).map(converter::convert);
}
}
}

210
src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java Normal file
View File

@@ -0,0 +1,210 @@
/*
* Copyright 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.data.repository.util;
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.Descriptor;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import lombok.experimental.UtilityClass;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Completable;
import rx.Observable;
import rx.Single;
/**
* Utility class to expose details about reactive wrapper types. This class exposes whether a reactive wrapper is
* supported in general and whether a particular type is suitable for no-value/single-value/multi-value usage.
* <p>
* Supported types are discovered by their availability on the class path. This class is typically used to determine
* multiplicity and whether a reactive wrapper type is acceptable for a specific operation.
*
* @author Mark Paluch
* @since 2.0
* @see org.reactivestreams.Publisher
* @see rx.Single
* @see rx.Observable
* @see rx.Completable
* @see io.reactivex.Single
* @see io.reactivex.Maybe
* @see io.reactivex.Observable
* @see io.reactivex.Completable
* @see io.reactivex.Flowable
* @see Mono
* @see Flux
*/
@UtilityClass
public class ReactiveWrappers {
static final boolean PROJECT_REACTOR_PRESENT = ClassUtils.isPresent("reactor.core.publisher.Mono",
ReactiveWrappers.class.getClassLoader());
static final boolean RXJAVA1_PRESENT = ClassUtils.isPresent("rx.Completable",
ReactiveWrappers.class.getClassLoader());
static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Flowable",
ReactiveWrappers.class.getClassLoader());
private static final Map<Class<?>, Descriptor> REACTIVE_WRAPPERS;
static {
Map<Class<?>, Descriptor> reactiveWrappers = new LinkedHashMap<>(3);
if (RXJAVA1_PRESENT) {
reactiveWrappers.put(Single.class, new Descriptor(false, true, false));
reactiveWrappers.put(Completable.class, new Descriptor(false, true, true));
reactiveWrappers.put(Observable.class, new Descriptor(true, true, false));
}
if (RXJAVA2_PRESENT) {
reactiveWrappers.put(io.reactivex.Single.class, new Descriptor(false, true, false));
reactiveWrappers.put(io.reactivex.Maybe.class, new Descriptor(false, true, false));
reactiveWrappers.put(io.reactivex.Completable.class, new Descriptor(false, true, true));
reactiveWrappers.put(io.reactivex.Flowable.class, new Descriptor(true, true, false));
reactiveWrappers.put(io.reactivex.Observable.class, new Descriptor(true, true, false));
}
if (PROJECT_REACTOR_PRESENT) {
reactiveWrappers.put(Mono.class, new Descriptor(false, true, false));
reactiveWrappers.put(Flux.class, new Descriptor(true, true, true));
reactiveWrappers.put(Publisher.class, new Descriptor(true, true, true));
}
REACTIVE_WRAPPERS = Collections.unmodifiableMap(reactiveWrappers);
}
/**
* Returns {@literal true} if reactive support is available. More specifically, whether RxJava1/2 or Project Reactor
* libraries are on the class path.
*
* @return {@literal true} if reactive support is available.
*/
public static boolean isAvailable() {
return RXJAVA1_PRESENT || RXJAVA2_PRESENT || PROJECT_REACTOR_PRESENT;
}
/**
* Returns {@literal true} if the {@code type} is a supported reactive wrapper type.
*
* @param type must not be {@literal null}.
* @return {@literal true} if the {@code type} is a supported reactive wrapper type.
*/
public static boolean supports(Class<?> type) {
return isNoValueType(type) || isSingleValueType(type) || isMultiValueType(type);
}
/**
* Returns {@literal true} if {@code type} is a reactive wrapper type that contains no value.
*
* @param type must not be {@literal null}.
* @return {@literal true} if {@code type} is a reactive wrapper type that contains no value.
*/
public static boolean isNoValueType(Class<?> type) {
Assert.notNull(type, "Class must not be null!");
return findDescriptor(type).map(Descriptor::isNoValue).orElse(false);
}
/**
* Returns {@literal true} if {@code type} is a reactive wrapper type for a single value.
*
* @param type must not be {@literal null}.
* @return {@literal true} if {@code type} is a reactive wrapper type for a single value.
*/
public static boolean isSingleValueType(Class<?> type) {
Assert.notNull(type, "Class must not be null!");
return findDescriptor(type).map((descriptor) -> !descriptor.isMultiValue() && !descriptor.isNoValue())
.orElse(false);
}
/**
* Returns {@literal true} if {@code type} is a reactive wrapper type supporting multiple values ({@code 0..N}
* elements).
*
* @param type must not be {@literal null}.
* @return {@literal true} if {@code type} is a reactive wrapper type supporting multiple values ({@code 0..N}
* elements).
*/
public static boolean isMultiValueType(Class<?> type) {
Assert.notNull(type, "Class must not be null!");
// Prevent single-types with a multi-hierarchy supertype to be reported as multi type
// See Mono implements Publisher
if (isSingleValueType(type)) {
return false;
}
return findDescriptor(type).map(Descriptor::isMultiValue).orElse(false);
}
/**
* Returns a collection of No-Value wrapper types.
*
* @return a collection of No-Value wrapper types.
*/
public static Collection<Class<?>> getNoValueTypes() {
return REACTIVE_WRAPPERS.entrySet().stream().filter(entry -> entry.getValue().isNoValue()).map(Entry::getKey)
.collect(Collectors.toList());
}
/**
* Returns a collection of Single-Value wrapper types.
*
* @return a collection of Single-Value wrapper types.
*/
public static Collection<Class<?>> getSingleValueTypes() {
return REACTIVE_WRAPPERS.entrySet().stream().filter(entry -> !entry.getValue().isMultiValue()).map(Entry::getKey)
.collect(Collectors.toList());
}
/**
* Returns a collection of Multi-Value wrapper types.
*
* @return a collection of Multi-Value wrapper types.
*/
public static Collection<Class<?>> getMultiValueTypes() {
return REACTIVE_WRAPPERS.entrySet().stream().filter(entry -> entry.getValue().isMultiValue()).map(Entry::getKey)
.collect(Collectors.toList());
}
private static Optional<Descriptor> findDescriptor(Class<?> rhsType) {
for (Class<?> type : REACTIVE_WRAPPERS.keySet()) {
if (org.springframework.util.ClassUtils.isAssignable(type, rhsType)) {
return Optional.ofNullable(REACTIVE_WRAPPERS.get(type));
}
}
return Optional.empty();
}
}