DATACMNS-836 - Add support for RxJava2.

Author: mp911de

pom.xml

@@ -18,6 +18,7 @@
 	<properties>
 		<scala>2.11.7</scala>
 		<xmlbeam>1.4.8</xmlbeam>
+		<rxjava>1.1.9</rxjava>
 	</properties>
 
 	<dependencies>
@@ -113,6 +114,13 @@
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>io.reactivex.rxjava2</groupId>
+			<artifactId>rxjava</artifactId>
+			<version>${rxjava2}</version>
+			<optional>true</optional>
+		</dependency>
+
 		<!--  Querydsl -->
 
 		<dependency>

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

@@ -47,17 +47,24 @@ public abstract class ReactiveWrapperConverters {
 
 	static {
 
+		if (RXJAVA1_PRESENT) {
+			REACTIVE_WRAPPERS.add(RxJava1SingleWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava1ObservableWrapper.INSTANCE);
+		}
+
+		if (RXJAVA2_PRESENT) {
+			REACTIVE_WRAPPERS.add(RxJava2SingleWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava2MaybeWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava2ObservableWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava2FlowableWrapper.INSTANCE);
+		}
+
 		if (PROJECT_REACTOR_PRESENT) {
 			REACTIVE_WRAPPERS.add(FluxWrapper.INSTANCE);
 			REACTIVE_WRAPPERS.add(MonoWrapper.INSTANCE);
 			REACTIVE_WRAPPERS.add(PublisherWrapper.INSTANCE);
 		}
 
-		if (RXJAVA1_PRESENT) {
-			REACTIVE_WRAPPERS.add(RxJava1SingleWrapper.INSTANCE);
-			REACTIVE_WRAPPERS.add(RxJava1ObservableWrapper.INSTANCE);
-		}
-
 		QueryExecutionConverters.registerConvertersIn(GENERIC_CONVERSION_SERVICE);
 	}
 
@@ -126,6 +133,7 @@ public static <T> T toWrapper(Object stream, Class<? extends T> expectedWrapperT
 	 * @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!");
@@ -252,6 +260,62 @@ public Observable<?> map(Object wrapper, Converter<Object, Object> converter) {
 		}
 	}
 
+	private static class RxJava2SingleWrapper extends AbstractReactiveWrapper<io.reactivex.Single<?>> {
+
+		static final RxJava2SingleWrapper INSTANCE = new RxJava2SingleWrapper();
+
+		private RxJava2SingleWrapper() {
+			super(io.reactivex.Single.class, Multiplicity.ONE);
+		}
+
+		@Override
+		public io.reactivex.Single<?> map(Object wrapper, Converter<Object, Object> converter) {
+			return ((io.reactivex.Single<?>) wrapper).map(converter::convert);
+		}
+	}
+
+	private static class RxJava2MaybeWrapper extends AbstractReactiveWrapper<io.reactivex.Maybe<?>> {
+
+		static final RxJava2MaybeWrapper INSTANCE = new RxJava2MaybeWrapper();
+
+		private RxJava2MaybeWrapper() {
+			super(io.reactivex.Maybe.class, Multiplicity.MANY);
+		}
+
+		@Override
+		public io.reactivex.Maybe<?> map(Object wrapper, Converter<Object, Object> converter) {
+			return ((io.reactivex.Maybe<?>) wrapper).map(converter::convert);
+		}
+	}
+
+	private static class RxJava2ObservableWrapper extends AbstractReactiveWrapper<io.reactivex.Observable<?>> {
+
+		static final RxJava2ObservableWrapper INSTANCE = new RxJava2ObservableWrapper();
+
+		private RxJava2ObservableWrapper() {
+			super(io.reactivex.Observable.class, Multiplicity.MANY);
+		}
+
+		@Override
+		public io.reactivex.Observable<?> map(Object wrapper, Converter<Object, Object> converter) {
+			return ((io.reactivex.Observable<?>) wrapper).map(converter::convert);
+		}
+	}
+
+	private static class RxJava2FlowableWrapper extends AbstractReactiveWrapper<io.reactivex.Flowable<?>> {
+
+		static final RxJava2FlowableWrapper INSTANCE = new RxJava2FlowableWrapper();
+
+		private RxJava2FlowableWrapper() {
+			super(io.reactivex.Flowable.class, Multiplicity.MANY);
+		}
+
+		@Override
+		public io.reactivex.Flowable<?> map(Object wrapper, Converter<Object, Object> converter) {
+			return ((io.reactivex.Flowable<?>) wrapper).map(converter::convert);
+		}
+	}
+
 	private enum Multiplicity {
 		ONE, MANY,
 	}

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

@@ -49,6 +49,9 @@ public class ReactiveWrappers {
 	public static final boolean RXJAVA1_PRESENT = ClassUtils.isPresent("rx.Single",
 			ReactiveWrappers.class.getClassLoader());
 
+	public static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Single",
+			ReactiveWrappers.class.getClassLoader());
+
 	private static final Set<Class<?>> SINGLE_TYPES;
 	private static final Set<Class<?>> MULTI_TYPES;
 
@@ -62,6 +65,13 @@ public class ReactiveWrappers {
 			multiTypes.add(getRxJava1ObservableClass());
 		}
 
+		if (RXJAVA2_PRESENT) {
+			singleTypes.add(getRxJava2SingleClass());
+			multiTypes.add(getRxJava2ObservableClass());
+			multiTypes.add(getRxJava2FlowableClass());
+			multiTypes.add(getRxJava2MaybeClass());
+		}
+
 		if (PROJECT_REACTOR_PRESENT) {
 			singleTypes.add(getReactorMonoClass());
 			multiTypes.add(getReactorFluxClass());
@@ -118,6 +128,22 @@ private static Class<?> getRxJava1ObservableClass() {
 		return Observable.class;
 	}
 
+	private static Class<?> getRxJava2SingleClass() {
+		return io.reactivex.Single.class;
+	}
+
+	private static Class<?> getRxJava2ObservableClass() {
+		return io.reactivex.Observable.class;
+	}
+
+	private static Class<?> getRxJava2FlowableClass() {
+		return io.reactivex.Flowable.class;
+	}
+
+	private static Class<?> getRxJava2MaybeClass() {
+		return io.reactivex.Maybe.class;
+	}
+
 	private static Class<?> getReactorMonoClass() {
 		return Mono.class;
 	}