From 474c9981dab4b4c193161415d25348cd426a0dda Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Sat, 30 Jul 2016 18:40:52 -0700 Subject: [PATCH] DATACMNS-836 - Provide wrapper conversion, ReactiveWrappers and RxJava 2 type conversion. --- pom.xml | 16 +- ...positoryConfigurationExtensionSupport.java | 3 +- .../ReactiveRepositoryInformation.java | 29 +- .../query/ReactiveWrapperConverters.java | 267 ---------- .../repository/query/ResultProcessor.java | 21 +- .../reactive/ReactiveCrudRepository.java | 2 +- .../ReactivePagingAndSortingRepository.java | 1 + .../reactive/RxJavaCrudRepository.java | 3 +- .../RxJavaPagingAndSortingRepository.java | 1 + .../util/QueryExecutionConverters.java | 498 ++++++++++++++---- .../util/ReactiveWrapperConverters.java | 285 ++++++++++ .../repository/util/ReactiveWrappers.java | 210 ++++++++ .../ReactiveWrapperConvertersUnitTests.java | 158 ------ .../QueryExecutionConvertersUnitTests.java | 10 + .../ReactiveWrapperConvertersUnitTests.java | 312 +++++++++++ .../util/ReactiveWrappersUnitTests.java | 90 ++++ 16 files changed, 1368 insertions(+), 538 deletions(-) delete mode 100644 src/main/java/org/springframework/data/repository/query/ReactiveWrapperConverters.java create mode 100644 src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java create mode 100644 src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java delete mode 100644 src/test/java/org/springframework/data/repository/query/ReactiveWrapperConvertersUnitTests.java create mode 100644 src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java create mode 100644 src/test/java/org/springframework/data/repository/util/ReactiveWrappersUnitTests.java diff --git a/pom.xml b/pom.xml index bec5ffa86..355d30a49 100644 --- a/pom.xml +++ b/pom.xml @@ -17,8 +17,8 @@ 2.11.7 - 2.5.0.M4 1.4.8 + 5.0.0.BUILD-SNAPSHOT @@ -114,6 +114,20 @@ true + + io.reactivex + rxjava-reactive-streams + ${rxjava-reactive-streams} + true + + + + io.reactivex.rxjava2 + rxjava + ${rxjava2} + true + + diff --git a/src/main/java/org/springframework/data/repository/config/RepositoryConfigurationExtensionSupport.java b/src/main/java/org/springframework/data/repository/config/RepositoryConfigurationExtensionSupport.java index 1596b67e4..6e1806908 100644 --- a/src/main/java/org/springframework/data/repository/config/RepositoryConfigurationExtensionSupport.java +++ b/src/main/java/org/springframework/data/repository/config/RepositoryConfigurationExtensionSupport.java @@ -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(); diff --git a/src/main/java/org/springframework/data/repository/core/support/ReactiveRepositoryInformation.java b/src/main/java/org/springframework/data/repository/core/support/ReactiveRepositoryInformation.java index 3d3eac6b6..da8a5bc0f 100644 --- a/src/main/java/org/springframework/data/repository/core/support/ReactiveRepositoryInformation.java +++ b/src/main/java/org/springframework/data/repository/core/support/ReactiveRepositoryInformation.java @@ -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, 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, 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, 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, Integer> { final Method declaredMethod; @@ -253,7 +266,5 @@ public class ReactiveRepositoryInformation extends DefaultRepositoryInformation return true; } - } - } diff --git a/src/main/java/org/springframework/data/repository/query/ReactiveWrapperConverters.java b/src/main/java/org/springframework/data/repository/query/ReactiveWrapperConverters.java deleted file mode 100644 index 6012f194b..000000000 --- a/src/main/java/org/springframework/data/repository/query/ReactiveWrapperConverters.java +++ /dev/null @@ -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> 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 toWrapper(Object stream, Class 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 map(Object stream, Converter 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> assignableStream(Class type) { - - Assert.notNull(type, "Type must not be null!"); - - return findWrapper(wrapper -> ClassUtils.isAssignable(wrapper.getWrapperClass(), type)); - } - - private static Optional> findWrapper( - Predicate> predicate) { - - return REACTIVE_WRAPPERS.stream().filter(predicate).findFirst(); - } - - private abstract static class AbstractReactiveWrapper { - - private final Class wrapperClass; - private final Multiplicity multiplicity; - - public AbstractReactiveWrapper(Class wrapperClass, Multiplicity multiplicity) { - this.wrapperClass = wrapperClass; - this.multiplicity = multiplicity; - } - - public Class getWrapperClass() { - return wrapperClass; - } - - public Multiplicity getMultiplicity() { - return multiplicity; - } - - public abstract Object map(Object wrapper, Converter converter); - } - - private static class MonoWrapper extends AbstractReactiveWrapper> { - - static final MonoWrapper INSTANCE = new MonoWrapper(); - - private MonoWrapper() { - super(Mono.class, Multiplicity.ONE); - } - - public Mono map(Object wrapper, Converter converter) { - return ((Mono) wrapper).map(converter::convert); - } - } - - private static class FluxWrapper extends AbstractReactiveWrapper> { - - static final FluxWrapper INSTANCE = new FluxWrapper(); - - private FluxWrapper() { - super(Flux.class, Multiplicity.MANY); - } - - public Flux map(Object wrapper, Converter converter) { - return ((Flux) wrapper).map(converter::convert); - } - } - - private static class PublisherWrapper extends AbstractReactiveWrapper> { - - static final PublisherWrapper INSTANCE = new PublisherWrapper(); - - public PublisherWrapper() { - super(Publisher.class, Multiplicity.MANY); - } - - @Override - public Publisher map(Object wrapper, Converter 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> { - - static final SingleWrapper INSTANCE = new SingleWrapper(); - - private SingleWrapper() { - super(Single.class, Multiplicity.ONE); - } - - @Override - public Single map(Object wrapper, Converter converter) { - return ((Single) wrapper).map(converter::convert); - } - } - - private static class ObservableWrapper extends AbstractReactiveWrapper> { - - static final ObservableWrapper INSTANCE = new ObservableWrapper(); - - private ObservableWrapper() { - super(Observable.class, Multiplicity.MANY); - } - - @Override - public Observable map(Object wrapper, Converter converter) { - return ((Observable) wrapper).map(converter::convert); - } - } - - private enum Multiplicity { - ONE, MANY, - } - -} diff --git a/src/main/java/org/springframework/data/repository/query/ResultProcessor.java b/src/main/java/org/springframework/data/repository/query/ResultProcessor.java index 39f97037e..0dfb32d19 100644 --- a/src/main/java/org/springframework/data/repository/query/ResultProcessor.java +++ b/src/main/java/org/springframework/data/repository/query/ResultProcessor.java @@ -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) */ diff --git a/src/main/java/org/springframework/data/repository/reactive/ReactiveCrudRepository.java b/src/main/java/org/springframework/data/repository/reactive/ReactiveCrudRepository.java index a15d331c3..f1af95c4e 100644 --- a/src/main/java/org/springframework/data/repository/reactive/ReactiveCrudRepository.java +++ b/src/main/java/org/springframework/data/repository/reactive/ReactiveCrudRepository.java @@ -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 extends Repository { diff --git a/src/main/java/org/springframework/data/repository/reactive/ReactivePagingAndSortingRepository.java b/src/main/java/org/springframework/data/repository/reactive/ReactivePagingAndSortingRepository.java index dd872bb51..15c94aa5b 100644 --- a/src/main/java/org/springframework/data/repository/reactive/ReactivePagingAndSortingRepository.java +++ b/src/main/java/org/springframework/data/repository/reactive/ReactivePagingAndSortingRepository.java @@ -31,6 +31,7 @@ import reactor.core.publisher.Mono; * and sorting abstraction. * * @author Mark Paluch + * @since 2.0 * @see Sort * @see Pageable * @see Mono diff --git a/src/main/java/org/springframework/data/repository/reactive/RxJavaCrudRepository.java b/src/main/java/org/springframework/data/repository/reactive/RxJavaCrudRepository.java index 795764706..a03886e12 100644 --- a/src/main/java/org/springframework/data/repository/reactive/RxJavaCrudRepository.java +++ b/src/main/java/org/springframework/data/repository/reactive/RxJavaCrudRepository.java @@ -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 */ diff --git a/src/main/java/org/springframework/data/repository/reactive/RxJavaPagingAndSortingRepository.java b/src/main/java/org/springframework/data/repository/reactive/RxJavaPagingAndSortingRepository.java index 3f6b52bf2..452e696dd 100644 --- a/src/main/java/org/springframework/data/repository/reactive/RxJavaPagingAndSortingRepository.java +++ b/src/main/java/org/springframework/data/repository/reactive/RxJavaPagingAndSortingRepository.java @@ -31,6 +31,7 @@ import rx.Single; * abstraction. * * @author Mark Paluch + * @since 2.0 * @see Sort * @see Pageable * @see Single diff --git a/src/main/java/org/springframework/data/repository/util/QueryExecutionConverters.java b/src/main/java/org/springframework/data/repository/util/QueryExecutionConverters.java index c12bf64b0..9e099eb05 100644 --- a/src/main/java/org/springframework/data/repository/util/QueryExecutionConverters.java +++ b/src/main/java/org/springframework/data/repository/util/QueryExecutionConverters.java @@ -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; *
  • {@code java.util.concurrent.Future}
    • *
  • {@code java.util.concurrent.CompletableFuture}
    • *
  • {@code org.springframework.util.concurrent.ListenableFuture<}
    • - *
  • {@code rx.Single}
    • - *
  • {@code rx.Observable}
    • - *
  • {@code rx.Completable}
    • - *
  • {@code reactor.core.publisher.Mono}
    • - *
  • {@code reactor.core.publisher.Flux}
    • - *
  • {@code org.reactivestreams.Publisher}
    • + *
  • Reactive wrappers supported by {@link ReactiveWrappers}
    • * * * @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> WRAPPER_TYPES = new HashSet>(); private static final Set> UNWRAPPER_TYPES = new HashSet>(); private static final Set> UNWRAPPERS = new HashSet>(); + 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, Single> { + public enum PublisherToRxJava1SingleConverter implements Converter, 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, Completable> { + public enum PublisherToRxJava1CompletableConverter implements Converter, 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, Observable> { + public enum PublisherToRxJava1ObservableConverter implements Converter, 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, Publisher> { + public enum RxJava1SingleToPublisherConverter implements Converter, 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, Mono> { + public enum RxJava1SingleToMonoConverter implements Converter, 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, Flux> { + public enum RxJava1SingleToFluxConverter implements Converter, 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> { + public enum RxJava1CompletableToPublisherConverter implements Converter> { 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> { + public enum RxJava1CompletableToMonoConverter implements Converter> { 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, Publisher> { + public enum RxJava1ObservableToPublisherConverter implements Converter, 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, Mono> { + public enum RxJava1ObservableToMonoConverter implements Converter, 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, Flux> { + public enum RxJava1ObservableToFluxConverter implements Converter, 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, 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, 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, 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, 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, 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, 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> { + + 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> { + + 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, Single> { + public enum RxJava1ObservableToSingleConverter implements Converter, Single> { INSTANCE; @@ -777,7 +1061,7 @@ public abstract class QueryExecutionConverters { * @author Mark Paluch * @author 2.0 */ - public enum SingleToObservableConverter implements Converter, Observable> { + public enum RxJava1SingleToObservableConverter implements Converter, 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.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.Observable> { + + INSTANCE; + + @Override + public io.reactivex.Observable convert(io.reactivex.Single source) { + return source.toObservable(); + } + } } diff --git a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java new file mode 100644 index 000000000..8f11c9ac5 --- /dev/null +++ b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java @@ -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}. + *

    + * 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> 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. + *

    + * NOTE: A reactive wrapper type might be supported in general by {@link ReactiveWrappers#supports(Class)} but not + * necessarily for conversion using this method. + *

    + * + * @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 toWrapper(Object stream, Class 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 map(Object stream, Converter 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 { + + Class getWrapperClass(); + + Object map(Object wrapper, Converter converter); + } + + private enum MonoWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return Mono.class; + } + + @Override + public Mono map(Object wrapper, Converter converter) { + return ((Mono) wrapper).map(converter::convert); + } + } + + private enum FluxWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return Flux.class; + } + + public Flux map(Object wrapper, Converter converter) { + return ((Flux) wrapper).map(converter::convert); + } + } + + private enum PublisherWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return Publisher.class; + } + + @Override + public Publisher map(Object wrapper, Converter 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> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return Single.class; + } + + @Override + public Single map(Object wrapper, Converter converter) { + return ((Single) wrapper).map(converter::convert); + } + } + + private enum RxJava1ObservableWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return Observable.class; + } + + @Override + public Observable map(Object wrapper, Converter converter) { + return ((Observable) wrapper).map(converter::convert); + } + } + + private enum RxJava2SingleWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return io.reactivex.Single.class; + } + + @Override + public io.reactivex.Single map(Object wrapper, Converter converter) { + return ((io.reactivex.Single) wrapper).map(converter::convert); + } + } + + private enum RxJava2MaybeWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return io.reactivex.Maybe.class; + } + + @Override + public io.reactivex.Maybe map(Object wrapper, Converter converter) { + return ((io.reactivex.Maybe) wrapper).map(converter::convert); + } + } + + private enum RxJava2ObservableWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return io.reactivex.Observable.class; + } + + @Override + public io.reactivex.Observable map(Object wrapper, Converter converter) { + return ((io.reactivex.Observable) wrapper).map(converter::convert); + } + } + + private enum RxJava2FlowableWrapper implements AbstractReactiveWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return io.reactivex.Flowable.class; + } + + @Override + public io.reactivex.Flowable map(Object wrapper, Converter converter) { + return ((io.reactivex.Flowable) wrapper).map(converter::convert); + } + } +} diff --git a/src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java b/src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java new file mode 100644 index 000000000..f4338e71b --- /dev/null +++ b/src/main/java/org/springframework/data/repository/util/ReactiveWrappers.java @@ -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. + *

    + * 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, Descriptor> REACTIVE_WRAPPERS; + + static { + + Map, 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> 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> 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> getMultiValueTypes() { + return REACTIVE_WRAPPERS.entrySet().stream().filter(entry -> entry.getValue().isMultiValue()).map(Entry::getKey) + .collect(Collectors.toList()); + } + + private static Optional 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(); + } +} diff --git a/src/test/java/org/springframework/data/repository/query/ReactiveWrapperConvertersUnitTests.java b/src/test/java/org/springframework/data/repository/query/ReactiveWrapperConvertersUnitTests.java deleted file mode 100644 index fc15b1cae..000000000 --- a/src/test/java/org/springframework/data/repository/query/ReactiveWrapperConvertersUnitTests.java +++ /dev/null @@ -1,158 +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 static org.assertj.core.api.AssertionsForClassTypes.*; - -import org.junit.Test; -import org.reactivestreams.Publisher; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import rx.Completable; -import rx.Observable; -import rx.Single; - -/** - * Unit tests for {@link ReactiveWrapperConverters}. - * - * @author Mark Paluch - */ -public class ReactiveWrapperConvertersUnitTests { - - /** - * @see DATACMNS-836 - */ - @Test - public void shouldSupportReactorTypes() { - - assertThat(ReactiveWrapperConverters.supports(Mono.class)).isTrue(); - assertThat(ReactiveWrapperConverters.supports(Flux.class)).isTrue(); - assertThat(ReactiveWrapperConverters.supports(Publisher.class)).isTrue(); - assertThat(ReactiveWrapperConverters.supports(Object.class)).isFalse(); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void shouldSupportRxJavaTypes() { - - assertThat(ReactiveWrapperConverters.supports(Single.class)).isTrue(); - assertThat(ReactiveWrapperConverters.supports(Observable.class)).isTrue(); - assertThat(ReactiveWrapperConverters.supports(Completable.class)).isFalse(); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void isSingleLikeShouldReportCorrectSingleTypes() { - - assertThat(ReactiveWrapperConverters.isSingleLike(Mono.class)).isTrue(); - assertThat(ReactiveWrapperConverters.isSingleLike(Flux.class)).isFalse(); - assertThat(ReactiveWrapperConverters.isSingleLike(Single.class)).isTrue(); - assertThat(ReactiveWrapperConverters.isSingleLike(Observable.class)).isFalse(); - assertThat(ReactiveWrapperConverters.isSingleLike(Publisher.class)).isFalse(); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void isCollectionLikeShouldReportCorrectCollectionTypes() { - - assertThat(ReactiveWrapperConverters.isCollectionLike(Mono.class)).isFalse(); - assertThat(ReactiveWrapperConverters.isCollectionLike(Flux.class)).isTrue(); - assertThat(ReactiveWrapperConverters.isCollectionLike(Single.class)).isFalse(); - assertThat(ReactiveWrapperConverters.isCollectionLike(Observable.class)).isTrue(); - assertThat(ReactiveWrapperConverters.isCollectionLike(Publisher.class)).isTrue(); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void toWrapperShouldCastMonoToMono() { - - Mono foo = Mono.just("foo"); - assertThat(ReactiveWrapperConverters.toWrapper(foo, Mono.class)).isSameAs(foo); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void toWrapperShouldConvertMonoToSingle() { - - Mono foo = Mono.just("foo"); - assertThat(ReactiveWrapperConverters.toWrapper(foo, Single.class)).isInstanceOf(Single.class); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void toWrapperShouldConvertMonoToFlux() { - - Mono foo = Mono.just("foo"); - assertThat(ReactiveWrapperConverters.toWrapper(foo, Flux.class)).isInstanceOf(Flux.class); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void shouldMapMono() { - - Mono foo = Mono.just("foo"); - Mono map = ReactiveWrapperConverters.map(foo, source -> 1L); - assertThat(map.block()).isEqualTo(1L); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void shouldMapFlux() { - - Flux foo = Flux.just("foo"); - Flux map = ReactiveWrapperConverters.map(foo, source -> 1L); - assertThat(map.next().block()).isEqualTo(1L); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void shouldMapSingle() { - - Single foo = Single.just("foo"); - Single map = ReactiveWrapperConverters.map(foo, source -> 1L); - assertThat(map.toBlocking().value()).isEqualTo(1L); - } - - /** - * @see DATACMNS-836 - */ - @Test - public void shouldMapObservable() { - - Observable foo = Observable.just("foo"); - Observable map = ReactiveWrapperConverters.map(foo, source -> 1L); - assertThat(map.toBlocking().first()).isEqualTo(1L); - } -} diff --git a/src/test/java/org/springframework/data/repository/util/QueryExecutionConvertersUnitTests.java b/src/test/java/org/springframework/data/repository/util/QueryExecutionConvertersUnitTests.java index 7ac2e1f97..02d6518e2 100644 --- a/src/test/java/org/springframework/data/repository/util/QueryExecutionConvertersUnitTests.java +++ b/src/test/java/org/springframework/data/repository/util/QueryExecutionConvertersUnitTests.java @@ -84,6 +84,11 @@ public class QueryExecutionConvertersUnitTests { assertThat(QueryExecutionConverters.supports(Single.class), is(true)); assertThat(QueryExecutionConverters.supports(Completable.class), is(true)); assertThat(QueryExecutionConverters.supports(Observable.class), is(true)); + assertThat(QueryExecutionConverters.supports(io.reactivex.Single.class), is(true)); + assertThat(QueryExecutionConverters.supports(io.reactivex.Maybe.class), is(true)); + assertThat(QueryExecutionConverters.supports(io.reactivex.Completable.class), is(true)); + assertThat(QueryExecutionConverters.supports(io.reactivex.Flowable.class), is(true)); + assertThat(QueryExecutionConverters.supports(io.reactivex.Observable.class), is(true)); } /** @@ -111,6 +116,11 @@ public class QueryExecutionConvertersUnitTests { assertThat(QueryExecutionConverters.supportsUnwrapping(Single.class), is(false)); assertThat(QueryExecutionConverters.supportsUnwrapping(Completable.class), is(false)); assertThat(QueryExecutionConverters.supportsUnwrapping(Observable.class), is(false)); + assertThat(QueryExecutionConverters.supportsUnwrapping(io.reactivex.Single.class), is(false)); + assertThat(QueryExecutionConverters.supportsUnwrapping(io.reactivex.Maybe.class), is(false)); + assertThat(QueryExecutionConverters.supportsUnwrapping(io.reactivex.Completable.class), is(false)); + assertThat(QueryExecutionConverters.supportsUnwrapping(io.reactivex.Flowable.class), is(false)); + assertThat(QueryExecutionConverters.supportsUnwrapping(io.reactivex.Observable.class), is(false)); } /** diff --git a/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java b/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java new file mode 100644 index 000000000..2b61c02db --- /dev/null +++ b/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java @@ -0,0 +1,312 @@ +/* + * 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 static org.assertj.core.api.AssertionsForClassTypes.*; + +import org.junit.Test; +import org.reactivestreams.Publisher; + +import io.reactivex.Flowable; +import org.springframework.data.repository.util.ReactiveWrapperConverters; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import rx.Completable; +import rx.Observable; +import rx.Single; + +/** + * Unit tests for {@link ReactiveWrapperConverters}. + * + * @author Mark Paluch + */ +public class ReactiveWrapperConvertersUnitTests { + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldSupportReactorTypes() { + + assertThat(ReactiveWrapperConverters.supports(Mono.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(Flux.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(Publisher.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(Object.class)).isFalse(); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldSupportRxJava1Types() { + + assertThat(ReactiveWrapperConverters.supports(Single.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(Observable.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(Completable.class)).isTrue(); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldSupportRxJava2Types() { + + assertThat(ReactiveWrapperConverters.supports(io.reactivex.Single.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(io.reactivex.Maybe.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(io.reactivex.Observable.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(io.reactivex.Flowable.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(io.reactivex.Completable.class)).isTrue(); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldCastMonoToMono() { + + Mono foo = Mono.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Mono.class)).isSameAs(foo); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertMonoToRxJava1Single() { + + Mono foo = Mono.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Single.class)).isInstanceOf(Single.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertMonoToRxJava2Single() { + + Mono foo = Mono.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, io.reactivex.Single.class)) + .isInstanceOf(io.reactivex.Single.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2SingleToMono() { + + io.reactivex.Single foo = io.reactivex.Single.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Mono.class)).isInstanceOf(Mono.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2SingleToPublisher() { + + io.reactivex.Single foo = io.reactivex.Single.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Publisher.class)).isInstanceOf(Publisher.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2MaybeToMono() { + + io.reactivex.Maybe foo = io.reactivex.Maybe.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Mono.class)).isInstanceOf(Mono.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2MaybeToFlux() { + + io.reactivex.Maybe foo = io.reactivex.Maybe.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Flux.class)).isInstanceOf(Flux.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2MaybeToPublisher() { + + io.reactivex.Maybe foo = io.reactivex.Maybe.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Publisher.class)).isInstanceOf(Publisher.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2FlowableToMono() { + + io.reactivex.Flowable foo = io.reactivex.Flowable.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Mono.class)).isInstanceOf(Mono.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2FlowableToFlux() { + + io.reactivex.Flowable foo = io.reactivex.Flowable.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Flux.class)).isInstanceOf(Flux.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldCastRxJava2FlowableToPublisher() { + + io.reactivex.Flowable foo = io.reactivex.Flowable.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Publisher.class)).isSameAs(foo); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2ObservableToMono() { + + io.reactivex.Observable foo = io.reactivex.Observable.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Mono.class)).isInstanceOf(Mono.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2ObservableToFlux() { + + io.reactivex.Observable foo = io.reactivex.Observable.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Flux.class)).isInstanceOf(Flux.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertRxJava2ObservableToPublisher() { + + io.reactivex.Observable foo = io.reactivex.Observable.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Publisher.class)).isInstanceOf(Publisher.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void toWrapperShouldConvertMonoToFlux() { + + Mono foo = Mono.just("foo"); + assertThat(ReactiveWrapperConverters.toWrapper(foo, Flux.class)).isInstanceOf(Flux.class); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapMono() { + + Mono foo = Mono.just("foo"); + Mono map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.block()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapFlux() { + + Flux foo = Flux.just("foo"); + Flux map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.next().block()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapRxJava1Single() { + + Single foo = Single.just("foo"); + Single map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.toBlocking().value()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapRxJava1Observable() { + + Observable foo = Observable.just("foo"); + Observable map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.toBlocking().first()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapRxJava2Single() { + + io.reactivex.Single foo = io.reactivex.Single.just("foo"); + io.reactivex.Single map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.blockingGet()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapRxJava2Maybe() { + + io.reactivex.Maybe foo = io.reactivex.Maybe.just("foo"); + io.reactivex.Maybe map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.toSingle().blockingGet()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapRxJava2Observable() { + + io.reactivex.Observable foo = io.reactivex.Observable.just("foo"); + io.reactivex.Observable map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.blockingFirst()).isEqualTo(1L); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void shouldMapRxJava2Flowable() { + + io.reactivex.Flowable foo = io.reactivex.Flowable.just("foo"); + io.reactivex.Flowable map = ReactiveWrapperConverters.map(foo, source -> 1L); + assertThat(map.blockingFirst()).isEqualTo(1L); + } +} diff --git a/src/test/java/org/springframework/data/repository/util/ReactiveWrappersUnitTests.java b/src/test/java/org/springframework/data/repository/util/ReactiveWrappersUnitTests.java new file mode 100644 index 000000000..81fd9f556 --- /dev/null +++ b/src/test/java/org/springframework/data/repository/util/ReactiveWrappersUnitTests.java @@ -0,0 +1,90 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; + +import io.reactivex.Completable; +import io.reactivex.Flowable; +import org.junit.Test; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import rx.Observable; +import rx.Single; + +/** + * Unit tests for {@link ReactiveWrappers}. + * + * @author Mark Paluch + */ +public class ReactiveWrappersUnitTests { + + /** + * @see DATACMNS-836 + */ + @Test + public void isSingleLikeShouldReportCorrectNoTypes() { + + assertThat(ReactiveWrappers.isNoValueType(Mono.class)).isFalse(); + assertThat(ReactiveWrappers.isNoValueType(Flux.class)).isTrue(); + assertThat(ReactiveWrappers.isNoValueType(Single.class)).isFalse(); + assertThat(ReactiveWrappers.isNoValueType(Completable.class)).isTrue(); + assertThat(ReactiveWrappers.isNoValueType(Observable.class)).isFalse(); + assertThat(ReactiveWrappers.isNoValueType(Publisher.class)).isTrue(); + assertThat(ReactiveWrappers.isNoValueType(io.reactivex.Single.class)).isFalse(); + assertThat(ReactiveWrappers.isNoValueType(io.reactivex.Maybe.class)).isFalse(); + assertThat(ReactiveWrappers.isNoValueType(Flowable.class)).isFalse(); + assertThat(ReactiveWrappers.isNoValueType(io.reactivex.Observable.class)).isFalse(); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void isSingleLikeShouldReportCorrectSingleTypes() { + + assertThat(ReactiveWrappers.isSingleValueType(Mono.class)).isTrue(); + assertThat(ReactiveWrappers.isSingleValueType(Flux.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(Single.class)).isTrue(); + assertThat(ReactiveWrappers.isSingleValueType(Completable.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(Observable.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(Publisher.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(io.reactivex.Single.class)).isTrue(); + assertThat(ReactiveWrappers.isSingleValueType(io.reactivex.Completable.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(io.reactivex.Maybe.class)).isTrue(); + assertThat(ReactiveWrappers.isSingleValueType(Flowable.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(io.reactivex.Observable.class)).isFalse(); + } + + /** + * @see DATACMNS-836 + */ + @Test + public void isCollectionLikeShouldReportCorrectCollectionTypes() { + + assertThat(ReactiveWrappers.isMultiValueType(Mono.class)).isFalse(); + assertThat(ReactiveWrappers.isMultiValueType(Flux.class)).isTrue(); + assertThat(ReactiveWrappers.isMultiValueType(Single.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(Completable.class)).isFalse(); + assertThat(ReactiveWrappers.isMultiValueType(Observable.class)).isTrue(); + assertThat(ReactiveWrappers.isMultiValueType(Publisher.class)).isTrue(); + assertThat(ReactiveWrappers.isMultiValueType(io.reactivex.Single.class)).isFalse(); + assertThat(ReactiveWrappers.isSingleValueType(io.reactivex.Completable.class)).isFalse(); + assertThat(ReactiveWrappers.isMultiValueType(Flowable.class)).isTrue(); + assertThat(ReactiveWrappers.isMultiValueType(io.reactivex.Observable.class)).isTrue(); + } +}