DATACMNS-1653 - Add support for RxJava 3.

We now include adapters and registry information for RxJava 3 types.

Original pull request: #444.

src/main/java/org/springframework/data/repository/reactive/RxJava3CrudRepository.java

@@ -0,0 +1,189 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.reactive;
+
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
+
+import org.springframework.data.repository.NoRepositoryBean;
+import org.springframework.data.repository.Repository;
+
+/**
+ * Interface for generic CRUD operations on a repository for a specific type. This repository follows reactive paradigms
+ * and uses RxJava 3 types.
+ *
+ * @author Mark Paluch
+ * @since 2.4
+ * @see Maybe
+ * @see Single
+ * @see Flowable
+ * @see Completable
+ */
+@NoRepositoryBean
+public interface RxJava3CrudRepository<T, ID> extends Repository<T, ID> {
+
+	/**
+	 * Saves a given entity. Use the returned instance for further operations as the save operation might have changed the
+	 * entity instance completely.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @return {@link Single} emitting the saved entity.
+	 * @throws IllegalArgumentException in case the given {@literal entity} is {@literal null}.
+	 */
+	<S extends T> Single<S> save(S entity);
+
+	/**
+	 * Saves all given entities.
+	 *
+	 * @param entities must not be {@literal null}.
+	 * @return {@link Flowable} emitting the saved entities.
+	 * @throws IllegalArgumentException in case the given {@link Iterable entities} or one of its entities is
+	 *           {@literal null}.
+	 */
+	<S extends T> Flowable<S> saveAll(Iterable<S> entities);
+
+	/**
+	 * Saves all given entities.
+	 *
+	 * @param entityStream must not be {@literal null}.
+	 * @return {@link Flowable} emitting the saved entities.
+	 * @throws IllegalArgumentException in case the given {@link Flowable entityStream} is {@literal null}.
+	 */
+	<S extends T> Flowable<S> saveAll(Flowable<S> entityStream);
+
+	/**
+	 * Retrieves an entity by its id.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return {@link Maybe} emitting the entity with the given id or {@link Maybe#empty()} if none found.
+	 * @throws IllegalArgumentException in case the given {@literal id} is {@literal null}.
+	 */
+	Maybe<T> findById(ID id);
+
+	/**
+	 * Retrieves an entity by its id supplied by a {@link Single}.
+	 *
+	 * @param id must not be {@literal null}. Uses the first emitted element to perform the find-query.
+	 * @return {@link Maybe} emitting the entity with the given id or {@link Maybe#empty()} if none found.
+	 * @throws IllegalArgumentException in case the given {@link Single id} is {@literal null}.
+	 */
+	Maybe<T> findById(Single<ID> id);
+
+	/**
+	 * Returns whether an entity with the given {@literal id} exists.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return {@link Single} emitting {@literal true} if an entity with the given id exists, {@literal false} otherwise.
+	 * @throws IllegalArgumentException in case the given {@literal id} is {@literal null}.
+	 */
+	Single<Boolean> existsById(ID id);
+
+	/**
+	 * Returns whether an entity with the given id, supplied by a {@link Single}, exists.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return {@link Single} emitting {@literal true} if an entity with the given id exists, {@literal false} otherwise.
+	 * @throws IllegalArgumentException in case the given {@link Single id} is {@literal null}.
+	 */
+	Single<Boolean> existsById(Single<ID> id);
+
+	/**
+	 * Returns all instances of the type.
+	 *
+	 * @return {@link Flowable} emitting all entities.
+	 */
+	Flowable<T> findAll();
+
+	/**
+	 * Returns all instances of the type {@code T} with the given IDs.
+	 * <p>
+	 * If some or all ids are not found, no entities are returned for these IDs.
+	 * <p>
+	 * Note that the order of elements in the result is not guaranteed.
+	 *
+	 * @param ids must not be {@literal null} nor contain any {@literal null} values.
+	 * @return {@link Flowable} emitting the found entities. The size can be equal or less than the number of given
+	 *         {@literal ids}.
+	 * @throws IllegalArgumentException in case the given {@link Iterable ids} or one of its items is {@literal null}.
+	 */
+	Flowable<T> findAllById(Iterable<ID> ids);
+
+	/**
+	 * Returns all instances of the type {@code T} with the given IDs supplied by a {@link Flowable}.
+	 * <p>
+	 * If some or all ids are not found, no entities are returned for these IDs.
+	 * <p>
+	 * Note that the order of elements in the result is not guaranteed.
+	 *
+	 * @param idStream must not be {@literal null}.
+	 * @return {@link Flowable} emitting the found entities.
+	 * @throws IllegalArgumentException in case the given {@link Flowable idStream} is {@literal null}.
+	 */
+	Flowable<T> findAllById(Flowable<ID> idStream);
+
+	/**
+	 * Returns the number of entities available.
+	 *
+	 * @return {@link Single} emitting the number of entities.
+	 */
+	Single<Long> count();
+
+	/**
+	 * Deletes the entity with the given id.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return {@link Completable} signaling when operation has completed.
+	 * @throws IllegalArgumentException in case the given {@literal id} is {@literal null}.
+	 */
+	Completable deleteById(ID id);
+
+	/**
+	 * Deletes a given entity.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @return {@link Completable} signaling when operation has completed.
+	 * @throws IllegalArgumentException in case the given entity is {@literal null}.
+	 */
+	Completable delete(T entity);
+
+	/**
+	 * Deletes the given entities.
+	 *
+	 * @param entities must not be {@literal null}.
+	 * @return {@link Completable} signaling when operation has completed.
+	 * @throws IllegalArgumentException in case the given {@link Iterable entities} or one of its entities is
+	 *           {@literal null}.
+	 */
+	Completable deleteAll(Iterable<? extends T> entities);
+
+	/**
+	 * Deletes the given entities supplied by a {@link Flowable}.
+	 *
+	 * @param entityStream must not be {@literal null}.
+	 * @return {@link Completable} signaling when operation has completed.
+	 * @throws IllegalArgumentException in case the given {@link Flowable entityStream} is {@literal null}.
+	 */
+	Completable deleteAll(Flowable<? extends T> entityStream);
+
+	/**
+	 * Deletes all entities managed by the repository.
+	 *
+	 * @return {@link Completable} signaling when operation has completed.
+	 */
+	Completable deleteAll();
+}

src/main/java/org/springframework/data/repository/reactive/RxJava3SortingRepository.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.reactive;
+
+import io.reactivex.rxjava3.core.Flowable;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.repository.NoRepositoryBean;
+
+/**
+ * Extension of {@link RxJava3CrudRepository} to provide additional methods to retrieve entities using the sorting
+ * abstraction.
+ *
+ * @author Mark Paluch
+ * @since 2.4
+ * @see Sort
+ * @see Flowable
+ * @see RxJava2CrudRepository
+ */
+@NoRepositoryBean
+public interface RxJava3SortingRepository<T, ID> extends RxJava3CrudRepository<T, ID> {
+
+	/**
+	 * Returns all entities sorted by the given options.
+	 *
+	 * @param sort must not be {@literal null}.
+	 * @return all entities sorted by the given options.
+	 */
+	Flowable<T> findAll(Sort sort);
+}

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

@@ -52,6 +52,8 @@ import org.springframework.util.ClassUtils;
  * <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.
+ * <p>
+ * <strong>Note:</strong> As of Spring Data 2.4, support for RxJava 1.x is deprecated in favor of RxJava 2 and 3.
  *
  * @author Mark Paluch
  * @author Oliver Gierke
@@ -80,6 +82,14 @@ public abstract class ReactiveWrapperConverters {
 			REACTIVE_WRAPPERS.add(RxJava2FlowableWrapper.INSTANCE);
 		}
 
+		if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA3)) {
+
+			REACTIVE_WRAPPERS.add(RxJava3SingleWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava3MaybeWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava3ObservableWrapper.INSTANCE);
+			REACTIVE_WRAPPERS.add(RxJava3FlowableWrapper.INSTANCE);
+		}
+
 		if (ReactiveWrappers.isAvailable(ReactiveLibrary.PROJECT_REACTOR)) {
 
 			REACTIVE_WRAPPERS.add(FluxWrapper.INSTANCE);
@@ -278,6 +288,10 @@ public abstract class ReactiveWrapperConverters {
 		}
 	}
 
+	// -------------------------------------------------------------------------
+	// RxJava 1 converters
+	// -------------------------------------------------------------------------
+
 	/**
 	 * Wrapper for RxJava 1's {@link Single}.
 	 */
@@ -314,6 +328,10 @@ public abstract class ReactiveWrapperConverters {
 		}
 	}
 
+	// -------------------------------------------------------------------------
+	// RxJava 2 converters
+	// -------------------------------------------------------------------------
+
 	/**
 	 * Wrapper for RxJava 2's {@link io.reactivex.Single}.
 	 */
@@ -386,6 +404,82 @@ public abstract class ReactiveWrapperConverters {
 		}
 	}
 
+	// -------------------------------------------------------------------------
+	// RxJava 3 converters
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Wrapper for RxJava 3's {@link io.reactivex.rxjava3.core.Single}.
+	 */
+	private static enum RxJava3SingleWrapper implements ReactiveTypeWrapper<io.reactivex.rxjava3.core.Single<?>> {
+
+		INSTANCE;
+
+		@Override
+		public Class<? super io.reactivex.rxjava3.core.Single<?>> getWrapperClass() {
+			return io.reactivex.rxjava3.core.Single.class;
+		}
+
+		@Override
+		public io.reactivex.rxjava3.core.Single<?> map(Object wrapper, Function<Object, Object> function) {
+			return ((io.reactivex.rxjava3.core.Single<?>) wrapper).map(function::apply);
+		}
+	}
+
+	/**
+	 * Wrapper for RxJava 3's {@link io.reactivex.rxjava3.core.Maybe}.
+	 */
+	private static enum RxJava3MaybeWrapper implements ReactiveTypeWrapper<io.reactivex.rxjava3.core.Maybe<?>> {
+
+		INSTANCE;
+
+		@Override
+		public Class<? super io.reactivex.rxjava3.core.Maybe<?>> getWrapperClass() {
+			return io.reactivex.rxjava3.core.Maybe.class;
+		}
+
+		@Override
+		public io.reactivex.rxjava3.core.Maybe<?> map(Object wrapper, Function<Object, Object> function) {
+			return ((io.reactivex.rxjava3.core.Maybe<?>) wrapper).map(function::apply);
+		}
+	}
+
+	/**
+	 * Wrapper for RxJava 3's {@link io.reactivex.rxjava3.core.Observable}.
+	 */
+	private static enum RxJava3ObservableWrapper implements ReactiveTypeWrapper<io.reactivex.rxjava3.core.Observable<?>> {
+
+		INSTANCE;
+
+		@Override
+		public Class<? super io.reactivex.rxjava3.core.Observable<?>> getWrapperClass() {
+			return io.reactivex.rxjava3.core.Observable.class;
+		}
+
+		@Override
+		public io.reactivex.rxjava3.core.Observable<?> map(Object wrapper, Function<Object, Object> function) {
+			return ((io.reactivex.rxjava3.core.Observable<?>) wrapper).map(function::apply);
+		}
+	}
+
+	/**
+	 * Wrapper for RxJava 3's {@link io.reactivex.rxjava3.core.Flowable}.
+	 */
+	private static enum RxJava3FlowableWrapper implements ReactiveTypeWrapper<io.reactivex.rxjava3.core.Flowable<?>> {
+
+		INSTANCE;
+
+		@Override
+		public Class<? super io.reactivex.rxjava3.core.Flowable<?>> getWrapperClass() {
+			return io.reactivex.rxjava3.core.Flowable.class;
+		}
+
+		@Override
+		public io.reactivex.rxjava3.core.Flowable<?> map(Object wrapper, Function<Object, Object> function) {
+			return ((io.reactivex.rxjava3.core.Flowable<?>) wrapper).map(function::apply);
+		}
+	}
+
 	// -------------------------------------------------------------------------
 	// ReactiveStreams converters
 	// -------------------------------------------------------------------------

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

@@ -37,6 +37,8 @@ import org.springframework.util.ClassUtils;
  * <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.
+ * <p>
+ * <strong>Note:</strong> As of Spring Data 2.4, support for RxJava 1.x is deprecated in favor of RxJava 2 and 3.
  *
  * @author Mark Paluch
  * @author Christoph Strobl
@@ -51,6 +53,11 @@ import org.springframework.util.ClassUtils;
  * @see io.reactivex.Observable
  * @see io.reactivex.Completable
  * @see io.reactivex.Flowable
+ * @see io.reactivex.rxjava3.core.Single
+ * @see io.reactivex.rxjava3.core.Maybe
+ * @see io.reactivex.rxjava3.core.Observable
+ * @see io.reactivex.rxjava3.core.Completable
+ * @see io.reactivex.rxjava3.core.Flowable
  * @see Mono
  * @see Flux
  */
@@ -59,12 +66,16 @@ public abstract class ReactiveWrappers {
 	private static final boolean PROJECT_REACTOR_PRESENT = ClassUtils.isPresent("reactor.core.publisher.Mono",
 			ReactiveWrappers.class.getClassLoader());
 
+	@Deprecated
 	private static final boolean RXJAVA1_PRESENT = ClassUtils.isPresent("rx.Completable",
 			ReactiveWrappers.class.getClassLoader());
 
 	private static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Flowable",
 			ReactiveWrappers.class.getClassLoader());
 
+	private static final boolean RXJAVA3_PRESENT = ClassUtils.isPresent("io.reactivex.rxjava3.core.Flowable",
+			ReactiveWrappers.class.getClassLoader());
+
 	private static final boolean KOTLIN_COROUTINES_PRESENT = ClassUtils.isPresent("kotlinx.coroutines.flow.Flow",
 			ReactiveWrappers.class.getClassLoader())
 			&& ClassUtils.isPresent("kotlinx.coroutines.reactive.ReactiveFlowKt", ReactiveWrappers.class.getClassLoader())
@@ -77,8 +88,15 @@ public abstract class ReactiveWrappers {
 	 *
 	 * @author Mark Paluch
 	 */
-	public static enum ReactiveLibrary {
-		PROJECT_REACTOR, RXJAVA1, RXJAVA2, KOTLIN_COROUTINES;
+	public enum ReactiveLibrary {
+
+		PROJECT_REACTOR,
+
+		/**
+		 * @deprecated since 2.4, use RxJava 2 or 3 instead.
+		 */
+		@Deprecated
+		RXJAVA1, RXJAVA2, RXJAVA3, KOTLIN_COROUTINES;
 	}
 
 	/**
@@ -108,6 +126,8 @@ public abstract class ReactiveWrappers {
 				return RXJAVA1_PRESENT;
 			case RXJAVA2:
 				return RXJAVA2_PRESENT;
+			case RXJAVA3:
+				return RXJAVA3_PRESENT;
 			case KOTLIN_COROUTINES:
 				return PROJECT_REACTOR_PRESENT && KOTLIN_COROUTINES_PRESENT;
 			default: