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

DATACMNS-836 - Cleanups.

Browse Source 
- Rename RxJava...Repository to RxJava1...Repository
- Use Completable and Observable instead of Single for results without values/optional values
- Remove reactive paging for now as it does not really fit reactive data streaming
- Expose ReactiveWrappers.isAvailable(ReactiveLibrary) method to query library availability
This commit is contained in:
Mark Paluch
2016-11-07 10:09:51 +01:00
committed by Oliver Gierke
parent 474c9981da
commit dd2fe0a34d
9 changed files with 119 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/main/java/org/springframework/data/repository/reactive/ReactivePagingAndSortingRepository.java → src/main/java/org/springframework/data/repository/reactive/ReactiveSortingRepository.java
View File

@@ -18,8 +18,6 @@ package org.springframework.data.repository.reactive;
import java.io.Serializable;
import org.reactivestreams.Publisher;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.repository.NoRepositoryBean;
@@ -27,18 +25,17 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Extension of {@link ReactiveCrudRepository} to provide additional methods to retrieve entities using the pagination
* and sorting abstraction.
* Extension of {@link ReactiveCrudRepository} to provide additional methods to retrieve entities using the sorting
* abstraction.
*
* @author Mark Paluch
* @since 2.0
* @see Sort
* @see Pageable
* @see Mono
* @see Flux
*/
@NoRepositoryBean
public interface ReactivePagingAndSortingRepository<T, ID extends Serializable> extends ReactiveCrudRepository<T, ID> {
public interface ReactiveSortingRepository<T, ID extends Serializable> extends ReactiveCrudRepository<T, ID> {
/*
* (non-Javadoc)
@@ -68,12 +65,4 @@ public interface ReactivePagingAndSortingRepository<T, ID extends Serializable>
* @return all entities sorted by the given options
*/
Flux<T> findAll(Sort sort);
/**
* Returns a {@link Page} of entities meeting the paging restriction provided in the {@code Pageable} object.
*
* @param pageable
* @return a page of entities
*/
Mono<Page<T>> findAll(Pageable pageable);
}

17
src/main/java/org/springframework/data/repository/reactive/RxJavaCrudRepository.java → src/main/java/org/springframework/data/repository/reactive/RxJava1CrudRepository.java
View File

@@ -21,6 +21,7 @@ import org.reactivestreams.Publisher;
import org.springframework.data.repository.NoRepositoryBean;
import org.springframework.data.repository.Repository;
import rx.Completable;
import rx.Observable;
import rx.Single;
@@ -34,7 +35,7 @@ import rx.Single;
* @see Observable
*/
@NoRepositoryBean
public interface RxJavaCrudRepository<T, ID extends Serializable> extends Repository<T, ID> {
public interface RxJava1CrudRepository<T, ID extends Serializable> extends Repository<T, ID> {
/**
* Saves a given entity. Use the returned instance for further operations as the save operation might have changed the
@@ -70,7 +71,7 @@ public interface RxJavaCrudRepository<T, ID extends Serializable> extends Reposi
* @return the entity with the given id or {@literal null} if none found
* @throws IllegalArgumentException if {@code id} is {@literal null}
*/
Single<T> findOne(ID id);
Observable<T> findOne(ID id);
/**
* Retrieves an entity by its id supplied by a {@link Single}.
@@ -79,7 +80,7 @@ public interface RxJavaCrudRepository<T, ID extends Serializable> extends Reposi
* @return the entity with the given id or {@literal null} if none found
* @throws IllegalArgumentException if {@code id} is {@literal null}
*/
Single<T> findOne(Single<ID> id);
Observable<T> findOne(Single<ID> id);
/**
* Returns whether an entity with the given id exists.
@@ -135,7 +136,7 @@ public interface RxJavaCrudRepository<T, ID extends Serializable> extends Reposi
* @param id must not be {@literal null}.
* @throws IllegalArgumentException in case the given {@code id} is {@literal null}
*/
Single<Void> delete(ID id);
Completable delete(ID id);
/**
* Deletes a given entity.
@@ -143,7 +144,7 @@ public interface RxJavaCrudRepository<T, ID extends Serializable> extends Reposi
* @param entity
* @throws IllegalArgumentException in case the given entity is {@literal null}.
*/
Single<Void> delete(T entity);
Completable delete(T entity);
/**
* Deletes the given entities.
@@ -151,7 +152,7 @@ public interface RxJavaCrudRepository<T, ID extends Serializable> extends Reposi
* @param entities
* @throws IllegalArgumentException in case the given {@link Iterable} is {@literal null}.
*/
Single<Void> delete(Iterable<? extends T> entities);
Completable delete(Iterable<? extends T> entities);
/**
* Deletes the given entities.
@@ -159,10 +160,10 @@ public interface RxJavaCrudRepository<T, ID extends Serializable> extends Reposi
* @param entityStream
* @throws IllegalArgumentException in case the given {@link Publisher} is {@literal null}.
*/
Single<Void> delete(Observable<? extends T> entityStream);
Completable delete(Observable<? extends T> entityStream);
/**
* Deletes all entities managed by the repository.
*/
Single<Void> deleteAll();
Completable deleteAll();
}

18
src/main/java/org/springframework/data/repository/reactive/RxJavaPagingAndSortingRepository.java → src/main/java/org/springframework/data/repository/reactive/RxJava1SortingRepository.java
View File

@@ -17,9 +17,6 @@ package org.springframework.data.repository.reactive;
import java.io.Serializable;
import org.reactivestreams.Publisher;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.repository.NoRepositoryBean;
@@ -27,19 +24,18 @@ import rx.Observable;
import rx.Single;
/**
* Extension of {@link RxJavaCrudRepository} to provide additional methods to retrieve entities using the pagination and sorting
* Extension of {@link RxJava1CrudRepository} to provide additional methods to retrieve entities using the sorting
* abstraction.
*
* @author Mark Paluch
* @since 2.0
* @see Sort
* @see Pageable
* @see Single
* @see Observable
* @see RxJavaCrudRepository
* @see RxJava1CrudRepository
*/
@NoRepositoryBean
public interface RxJavaPagingAndSortingRepository<T, ID extends Serializable> extends RxJavaCrudRepository<T, ID> {
public interface RxJava1SortingRepository<T, ID extends Serializable> extends RxJava1CrudRepository<T, ID> {
/*
* (non-Javadoc)
@@ -69,12 +65,4 @@ public interface RxJavaPagingAndSortingRepository<T, ID extends Serializable> ex
* @return all entities sorted by the given options
*/
Observable<T> findAll(Sort sort);
/**
* Returns a {@link Page} of entities meeting the paging restriction provided in the {@code Pageable} object.
*
* @param pageable
* @return a page of entities
*/
Single<Page<T>> findAll(Pageable pageable);
}

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

@@ -28,6 +28,7 @@ import org.springframework.core.convert.TypeDescriptor;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.converter.GenericConverter;
import org.springframework.core.convert.support.ConfigurableConversionService;
import org.springframework.data.repository.util.ReactiveWrappers.ReactiveLibrary;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -189,7 +190,7 @@ public abstract class QueryExecutionConverters {
if (ReactiveWrappers.isAvailable()) {
if (ReactiveWrappers.RXJAVA1_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA1)) {
conversionService.addConverter(PublisherToRxJava1CompletableConverter.INSTANCE);
conversionService.addConverter(RxJava1CompletableToPublisherConverter.INSTANCE);
@@ -206,7 +207,7 @@ public abstract class QueryExecutionConverters {
conversionService.addConverter(RxJava1ObservableToFluxConverter.INSTANCE);
}
if (ReactiveWrappers.RXJAVA2_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA2)) {
conversionService.addConverter(PublisherToRxJava2CompletableConverter.INSTANCE);
conversionService.addConverter(RxJava2CompletableToPublisherConverter.INSTANCE);
@@ -231,19 +232,20 @@ public abstract class QueryExecutionConverters {
conversionService.addConverter(RxJava2MaybeToFluxConverter.INSTANCE);
}
if (ReactiveWrappers.PROJECT_REACTOR_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.PROJECT_REACTOR)) {
conversionService.addConverter(PublisherToMonoConverter.INSTANCE);
conversionService.addConverter(PublisherToFluxConverter.INSTANCE);
}
if (ReactiveWrappers.RXJAVA1_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA1)) {
conversionService.addConverter(RxJava1SingleToObservableConverter.INSTANCE);
conversionService.addConverter(RxJava1ObservableToSingleConverter.INSTANCE);
}
if (ReactiveWrappers.RXJAVA2_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA2)) {
conversionService.addConverter(RxJava2SingleToObservableConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToSingleConverter.INSTANCE);
conversionService.addConverter(RxJava2ObservableToMaybeConverter.INSTANCE);
}
}
}
@@ -1084,7 +1086,24 @@ public abstract class QueryExecutionConverters {
@Override
public io.reactivex.Single<?> convert(io.reactivex.Observable<?> source) {
return source.singleElement().toSingle();
return source.singleOrError();
}
}
/**
* A {@link Converter} to convert a {@link Observable} to {@link Maybe}.
*
* @author Mark Paluch
* @author 2.0
*/
public enum RxJava2ObservableToMaybeConverter
implements Converter<io.reactivex.Observable<?>, io.reactivex.Maybe<?>> {
INSTANCE;
@Override
public io.reactivex.Maybe<?> convert(io.reactivex.Observable<?> source) {
return source.singleElement();
}
}

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

@@ -22,6 +22,7 @@ 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.data.repository.util.ReactiveWrappers.ReactiveLibrary;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -52,13 +53,13 @@ public class ReactiveWrapperConverters {
static {
if (ReactiveWrappers.RXJAVA1_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA1)) {
REACTIVE_WRAPPERS.add(RxJava1SingleWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(RxJava1ObservableWrapper.INSTANCE);
}
if (ReactiveWrappers.RXJAVA2_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.RXJAVA2)) {
REACTIVE_WRAPPERS.add(RxJava2SingleWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(RxJava2MaybeWrapper.INSTANCE);
@@ -66,7 +67,7 @@ public class ReactiveWrapperConverters {
REACTIVE_WRAPPERS.add(RxJava2FlowableWrapper.INSTANCE);
}
if (ReactiveWrappers.PROJECT_REACTOR_PRESENT) {
if (ReactiveWrappers.isAvailable(ReactiveLibrary.PROJECT_REACTOR)) {
REACTIVE_WRAPPERS.add(FluxWrapper.INSTANCE);
REACTIVE_WRAPPERS.add(MonoWrapper.INSTANCE);

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

@@ -59,13 +59,13 @@ import rx.Single;
@UtilityClass
public class ReactiveWrappers {
static final boolean PROJECT_REACTOR_PRESENT = ClassUtils.isPresent("reactor.core.publisher.Mono",
private static final boolean PROJECT_REACTOR_PRESENT = ClassUtils.isPresent("reactor.core.publisher.Mono",
ReactiveWrappers.class.getClassLoader());
static final boolean RXJAVA1_PRESENT = ClassUtils.isPresent("rx.Completable",
private static final boolean RXJAVA1_PRESENT = ClassUtils.isPresent("rx.Completable",
ReactiveWrappers.class.getClassLoader());
static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Flowable",
private static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Flowable",
ReactiveWrappers.class.getClassLoader());
private static final Map<Class<?>, Descriptor> REACTIVE_WRAPPERS;
@@ -107,7 +107,30 @@ public class ReactiveWrappers {
* @return {@literal true} if reactive support is available.
*/
public static boolean isAvailable() {
return RXJAVA1_PRESENT || RXJAVA2_PRESENT || PROJECT_REACTOR_PRESENT;
return isAvailable(ReactiveLibrary.PROJECT_REACTOR) || isAvailable(ReactiveLibrary.RXJAVA1)
|| isAvailable(ReactiveLibrary.RXJAVA2);
}
/**
* Returns {@literal true} if the {@link ReactiveLibrary} is available.
*
* @param reactiveLibrary must not be {@literal null}.
* @return {@literal true} if the {@link ReactiveLibrary} is available.
*/
public static boolean isAvailable(ReactiveLibrary reactiveLibrary) {
Assert.notNull(reactiveLibrary, "ReactiveLibrary must not be null!");
switch (reactiveLibrary) {
case PROJECT_REACTOR:
return PROJECT_REACTOR_PRESENT;
case RXJAVA1:
return RXJAVA1_PRESENT;
case RXJAVA2:
return RXJAVA2_PRESENT;
}
throw new IllegalArgumentException(String.format("ReactiveLibrary %s not supported", reactiveLibrary));
}
/**
@@ -207,4 +230,13 @@ public class ReactiveWrappers {
}
return Optional.empty();
}
/**
* Enumeration of supported reactive libraries.
*
* @author Mark Paluch
*/
enum ReactiveLibrary {
PROJECT_REACTOR, RXJAVA1, RXJAVA2;
}
}