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

Further refactoring of ReactiveAdapter/Registry

Browse Source 
Simplify getAdapterFrom/To into a single getAdapter method that looks
for an exact match by type first and then isAssignableFrom.

Also expose shortcut methods in ReactiveAdapter to minimize the need
for access to the ReactiveTypeDescriptor.

Issue: SPR-14902
This commit is contained in:
Rossen Stoyanchev
2016-11-28 12:43:22 -05:00
parent adb80f4099
commit 5651c2180e
11 changed files with 136 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-core/src/main/java/org/springframework/core/ReactiveAdapter.java
View File

@@ -70,6 +70,35 @@ public class ReactiveAdapter {
return this.descriptor;
}
/**
* A shortcut for {@code getDescriptor().getReactiveType()}.
*/
public Class<?> getReactiveType() {
return getDescriptor().getReactiveType();
}
/**
* A shortcut for {@code getDescriptor().isMultiValue()}.
*/
public boolean isMultiValue() {
return getDescriptor().isMultiValue();
}
/**
* A shortcut for {@code getDescriptor().supportsEmpty()}.
*/
public boolean supportsEmpty() {
return getDescriptor().supportsEmpty();
}
/**
* A shortcut for {@code getDescriptor().isNoValue()}.
*/
public boolean isNoValue() {
return getDescriptor().isNoValue();
}
/**
* Adapt the given instance to a Reactive Streams Publisher.
* @param source the source object to adapt from

36
spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java
View File

@@ -21,7 +21,6 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.function.Predicate;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Completable;
@@ -101,34 +100,31 @@ public class ReactiveAdapterRegistry {
}
/**
* Get the adapter to use to adapt from the given reactive type.
* Get the adapter for the given reactive type.
*/
public ReactiveAdapter getAdapterFrom(Class<?> reactiveType) {
return getAdapterFrom(reactiveType, null);
public ReactiveAdapter getAdapter(Class<?> reactiveType) {
return getAdapter(reactiveType, null);
}
/**
* Get the adapter to use to adapt from the given reactive type. Or if the
* "source" object is not {@code null} its actual type is used instead.
* Get the adapter for the given reactive type. Or if a "source" object is
* provided, its actual type is used instead.
* @param reactiveType the reactive type
* @param source an instance of the reactive type (i.e. to adapt from)
*/
public ReactiveAdapter getAdapterFrom(Class<?> reactiveType, Object source) {
public ReactiveAdapter getAdapter(Class<?> reactiveType, Object source) {
source = (source instanceof Optional ? ((Optional<?>) source).orElse(null) : source);
Class<?> clazz = (source != null ? source.getClass() : reactiveType);
return getAdapter(type -> type.isAssignableFrom(clazz));
}
/**
* Get the adapter for the given reactive type to adapt to.
*/
public ReactiveAdapter getAdapterTo(Class<?> reactiveType) {
return getAdapter(reactiveType::equals);
}
private ReactiveAdapter getAdapter(Predicate<Class<?>> predicate) {
return this.adapters.stream()
.filter(adapter -> predicate.test(adapter.getDescriptor().getReactiveType()))
.filter(adapter -> adapter.getReactiveType().equals(clazz))
.findFirst()
.orElse(null);
.orElseGet(() ->
this.adapters.stream()
.filter(adapter -> adapter.getReactiveType().isAssignableFrom(clazz))
.findFirst()
.orElse(null));
}
@@ -233,7 +229,7 @@ public class ReactiveAdapterRegistry {
@Override
public <T> Publisher<T> toPublisher(Object source) {
Publisher<T> publisher = super.toPublisher(source);
return (getDescriptor().isMultiValue() ? Flux.from(publisher) : Mono.from(publisher));
return (isMultiValue() ? Flux.from(publisher) : Mono.from(publisher));
}
}

96
spring-core/src/test/java/org/springframework/core/convert/support/ReactiveAdapterRegistryTests.java
View File

@@ -25,6 +25,7 @@ import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.Mono;
import rx.Completable;
import rx.Observable;
@@ -32,10 +33,13 @@ import rx.Single;
import org.springframework.core.ReactiveAdapter;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.ReactiveTypeDescriptor;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
/**
@@ -55,33 +59,55 @@ public class ReactiveAdapterRegistryTests {
@Test
public void getDefaultAdapters() throws Exception {
public void defaultAdapterRegistrations() throws Exception {
// Reactor
assertNotNull(getAdapterTo(Mono.class));
assertNotNull(getAdapterTo(Flux.class));
assertNotNull(getAdapter(Mono.class));
assertNotNull(getAdapter(Flux.class));
assertNotNull(getAdapterTo(Publisher.class));
assertNotNull(getAdapterTo(CompletableFuture.class));
// Publisher
assertNotNull(getAdapter(Publisher.class));
// Completable
assertNotNull(getAdapter(CompletableFuture.class));
// RxJava 1
assertNotNull(getAdapterTo(Observable.class));
assertNotNull(getAdapterTo(Single.class));
assertNotNull(getAdapterTo(Completable.class));
assertNotNull(getAdapter(Observable.class));
assertNotNull(getAdapter(Single.class));
assertNotNull(getAdapter(Completable.class));
// RxJava 2
assertNotNull(getAdapterTo(Flowable.class));
assertNotNull(getAdapterTo(io.reactivex.Observable.class));
assertNotNull(getAdapterTo(io.reactivex.Single.class));
assertNotNull(getAdapterTo(Maybe.class));
assertNotNull(getAdapterTo(io.reactivex.Completable.class));
assertNotNull(getAdapter(Flowable.class));
assertNotNull(getAdapter(io.reactivex.Observable.class));
assertNotNull(getAdapter(io.reactivex.Single.class));
assertNotNull(getAdapter(Maybe.class));
assertNotNull(getAdapter(io.reactivex.Completable.class));
}
@Test
public void getAdapterForReactiveSubType() throws Exception {
ReactiveAdapter adapter1 = getAdapter(Flux.class);
ReactiveAdapter adapter2 = getAdapter(FluxProcessor.class);
assertSame(adapter1, adapter2);
this.registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(FluxProcessor.class, FluxProcessor::empty),
o -> (FluxProcessor<?, ?>) o,
FluxProcessor::from);
ReactiveAdapter adapter3 = getAdapter(FluxProcessor.class);
assertNotNull(adapter3);
assertNotSame(adapter1, adapter3);
}
@Test
public void publisherToFlux() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flowable.fromIterable(sequence);
Object target = getAdapterTo(Flux.class).fromPublisher(source);
Object target = getAdapter(Flux.class).fromPublisher(source);
assertTrue(target instanceof Flux);
assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
}
@@ -91,7 +117,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void publisherToMono() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapterTo(Mono.class).fromPublisher(source);
Object target = getAdapter(Mono.class).fromPublisher(source);
assertTrue(target instanceof Mono);
assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
}
@@ -99,7 +125,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void publisherToCompletableFuture() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapterTo(CompletableFuture.class).fromPublisher(source);
Object target = getAdapter(CompletableFuture.class).fromPublisher(source);
assertTrue(target instanceof CompletableFuture);
assertEquals(new Integer(1), ((CompletableFuture<Integer>) target).get());
}
@@ -108,7 +134,7 @@ public class ReactiveAdapterRegistryTests {
public void publisherToRxObservable() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flowable.fromIterable(sequence);
Object target = getAdapterTo(rx.Observable.class).fromPublisher(source);
Object target = getAdapter(rx.Observable.class).fromPublisher(source);
assertTrue(target instanceof rx.Observable);
assertEquals(sequence, ((rx.Observable) target).toList().toBlocking().first());
}
@@ -116,7 +142,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void publisherToRxSingle() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1);
Object target = getAdapterTo(rx.Single.class).fromPublisher(source);
Object target = getAdapter(rx.Single.class).fromPublisher(source);
assertTrue(target instanceof rx.Single);
assertEquals(new Integer(1), ((rx.Single<Integer>) target).toBlocking().value());
}
@@ -124,7 +150,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void publisherToRxCompletable() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapterTo(rx.Completable.class).fromPublisher(source);
Object target = getAdapter(rx.Completable.class).fromPublisher(source);
assertTrue(target instanceof rx.Completable);
assertNull(((rx.Completable) target).get());
}
@@ -133,7 +159,7 @@ public class ReactiveAdapterRegistryTests {
public void publisherToReactivexFlowable() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapterTo(io.reactivex.Flowable.class).fromPublisher(source);
Object target = getAdapter(io.reactivex.Flowable.class).fromPublisher(source);
assertTrue(target instanceof io.reactivex.Flowable);
assertEquals(sequence, ((io.reactivex.Flowable) target).toList().blockingGet());
}
@@ -142,7 +168,7 @@ public class ReactiveAdapterRegistryTests {
public void publisherToReactivexObservable() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flowable.fromIterable(sequence);
Object target = getAdapterTo(io.reactivex.Observable.class).fromPublisher(source);
Object target = getAdapter(io.reactivex.Observable.class).fromPublisher(source);
assertTrue(target instanceof io.reactivex.Observable);
assertEquals(sequence, ((io.reactivex.Observable) target).toList().blockingGet());
}
@@ -150,7 +176,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void publisherToReactivexSingle() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1);
Object target = getAdapterTo(io.reactivex.Single.class).fromPublisher(source);
Object target = getAdapter(io.reactivex.Single.class).fromPublisher(source);
assertTrue(target instanceof io.reactivex.Single);
assertEquals(new Integer(1), ((io.reactivex.Single<Integer>) target).blockingGet());
}
@@ -158,7 +184,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void publisherToReactivexCompletable() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapterTo(io.reactivex.Completable.class).fromPublisher(source);
Object target = getAdapter(io.reactivex.Completable.class).fromPublisher(source);
assertTrue(target instanceof io.reactivex.Completable);
assertNull(((io.reactivex.Completable) target).blockingGet());
}
@@ -167,7 +193,7 @@ public class ReactiveAdapterRegistryTests {
public void rxObservableToPublisher() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = rx.Observable.from(sequence);
Object target = getAdapterFrom(rx.Observable.class).toPublisher(source);
Object target = getAdapter(rx.Observable.class).toPublisher(source);
assertTrue("Expected Flux Publisher: " + target.getClass().getName(), target instanceof Flux);
assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
}
@@ -175,7 +201,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void rxSingleToPublisher() throws Exception {
Object source = rx.Single.just(1);
Object target = getAdapterFrom(rx.Single.class).toPublisher(source);
Object target = getAdapter(rx.Single.class).toPublisher(source);
assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
}
@@ -183,7 +209,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void rxCompletableToPublisher() throws Exception {
Object source = rx.Completable.complete();
Object target = getAdapterFrom(rx.Completable.class).toPublisher(source);
Object target = getAdapter(rx.Completable.class).toPublisher(source);
assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
((Mono<Void>) target).blockMillis(1000);
}
@@ -192,7 +218,7 @@ public class ReactiveAdapterRegistryTests {
public void reactivexFlowableToPublisher() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.Flowable.fromIterable(sequence);
Object target = getAdapterFrom(io.reactivex.Flowable.class).toPublisher(source);
Object target = getAdapter(io.reactivex.Flowable.class).toPublisher(source);
assertTrue("Expected Flux Publisher: " + target.getClass().getName(), target instanceof Flux);
assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
}
@@ -201,7 +227,7 @@ public class ReactiveAdapterRegistryTests {
public void reactivexObservableToPublisher() throws Exception {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.Observable.fromIterable(sequence);
Object target = getAdapterFrom(io.reactivex.Observable.class).toPublisher(source);
Object target = getAdapter(io.reactivex.Observable.class).toPublisher(source);
assertTrue("Expected Flux Publisher: " + target.getClass().getName(), target instanceof Flux);
assertEquals(sequence, ((Flux<Integer>) target).collectList().blockMillis(1000));
}
@@ -209,7 +235,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void reactivexSingleToPublisher() throws Exception {
Object source = io.reactivex.Single.just(1);
Object target = getAdapterFrom(io.reactivex.Single.class).toPublisher(source);
Object target = getAdapter(io.reactivex.Single.class).toPublisher(source);
assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
}
@@ -217,7 +243,7 @@ public class ReactiveAdapterRegistryTests {
@Test
public void reactivexCompletableToPublisher() throws Exception {
Object source = io.reactivex.Completable.complete();
Object target = getAdapterFrom(io.reactivex.Completable.class).toPublisher(source);
Object target = getAdapter(io.reactivex.Completable.class).toPublisher(source);
assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
((Mono<Void>) target).blockMillis(1000);
}
@@ -226,18 +252,14 @@ public class ReactiveAdapterRegistryTests {
public void CompletableFutureToPublisher() throws Exception {
CompletableFuture<Integer> future = new CompletableFuture();
future.complete(1);
Object target = getAdapterFrom(CompletableFuture.class).toPublisher(future);
Object target = getAdapter(CompletableFuture.class).toPublisher(future);
assertTrue("Expected Mono Publisher: " + target.getClass().getName(), target instanceof Mono);
assertEquals(new Integer(1), ((Mono<Integer>) target).blockMillis(1000));
}
private ReactiveAdapter getAdapterTo(Class<?> reactiveType) {
return this.registry.getAdapterTo(reactiveType);
}
private ReactiveAdapter getAdapterFrom(Class<?> reactiveType) {
return this.registry.getAdapterFrom(reactiveType);
private ReactiveAdapter getAdapter(Class<?> reactiveType) {
return this.registry.getAdapter(reactiveType);
}
}