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

Refactor ReactiveAdapter hierarchy

Browse Source 
Collapse ReactiveAdapter hierarchy into a single class that simply
delegates to functions for converting to/from a Publisher.

A private ReactorAdapter extension automaticlaly wraps adapted,  "raw"
Publisher's as Flux or Mono depending on the semantics of the target
reactive type.

Issue: SPR-14902
This commit is contained in:
Rossen Stoyanchev
2016-11-24 18:30:05 -05:00
parent e563326357
commit 52096ab8b9
8 changed files with 340 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,55 +16,81 @@
package org.springframework.core;
import java.util.Optional;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.util.Assert;
/**
* Contract for adapting to and from {@link Flux} and {@link Mono}.
* Adapt a Reactive Streams {@link Publisher} to and from an async/reactive type
* such as {@code CompletableFuture}, an RxJava {@code Observable}, etc.
*
* <p>An adapter supports a specific adaptee type whose stream semantics
* can be checked via {@link #getDescriptor()}.
*
* <p>Use the {@link ReactiveAdapterRegistry} to obtain an adapter for a
* supported adaptee type or to register additional adapters.
* <p>Use the {@link ReactiveAdapterRegistry} to register reactive types and
* obtain adapters from.
*
* @author Rossen Stoyanchev
* @since 5.0
*/
public interface ReactiveAdapter {
public class ReactiveAdapter {
private final ReactiveTypeDescriptor descriptor;
private final Function<Object, Publisher<?>> toPublisherFunction;
private final Function<Publisher<?>, Object> fromPublisherFunction;
/**
* Return a descriptor with further information about the adaptee.
* Constructor for an adapter with functions to convert the target reactive
* or async type to and from a Reactive Streams Publisher.
* @param descriptor the reactive type descriptor
* @param toPublisherFunction adapter to a Publisher
* @param fromPublisherFunction adapter from a Publisher
*/
ReactiveTypeDescriptor getDescriptor();
public ReactiveAdapter(ReactiveTypeDescriptor descriptor,
Function<Object, Publisher<?>> toPublisherFunction,
Function<Publisher<?>, Object> fromPublisherFunction) {
Assert.notNull(descriptor, "'descriptor' is required");
Assert.notNull(toPublisherFunction, "'toPublisherFunction' is required");
Assert.notNull(fromPublisherFunction, "'fromPublisherFunction' is required");
this.descriptor = descriptor;
this.toPublisherFunction = toPublisherFunction;
this.fromPublisherFunction = fromPublisherFunction;
}
/**
* Adapt the given Object to a {@link Mono}
* @param source the source object to adapt
* @return the resulting {@link Mono} possibly empty
* Return the descriptor of the reactive type for the adapter.
*/
<T> Mono<T> toMono(Object source);
public ReactiveTypeDescriptor getDescriptor() {
return this.descriptor;
}
/**
* Adapt the given Object to a {@link Flux}.
* @param source the source object to adapt
* @return the resulting {@link Flux} possibly empty
* Adapt the given instance to a Reactive Streams Publisher.
* @param source the source object to adapt from
* @return the Publisher repesenting the adaptation
*/
<T> Flux<T> toFlux(Object source);
@SuppressWarnings("unchecked")
public <T> Publisher<T> toPublisher(Object source) {
source = (source instanceof Optional ? ((Optional<?>) source).orElse(null) : source);
if (source == null) {
source = getDescriptor().getEmptyValue();
}
return (Publisher<T>) this.toPublisherFunction.apply(source);
}
/**
* Adapt the given Object to a Publisher.
* @param source the source object to adapt
* @return the resulting {@link Mono} or {@link Flux} possibly empty
* Adapt from the given Reactive Streams Publisher.
* @param publisher the publisher to adapt from
* @return the reactive type instance representing the adapted publisher
*/
<T> Publisher<T> toPublisher(Object source);
/**
* Adapt the given Publisher to the target adaptee.
* @param publisher the publisher to adapt
* @return the resulting adaptee
*/
Object fromPublisher(Publisher<?> publisher);
public Object fromPublisher(Publisher<?> publisher) {
return (publisher != null ? this.fromPublisherFunction.apply(publisher) : null);
}
}

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

@@ -24,8 +24,10 @@ import java.util.function.Function;
import java.util.function.Predicate;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -68,21 +70,25 @@ public class ReactiveAdapterRegistry {
// Flux and Mono ahead of Publisher...
registerReactiveType(
ReactiveTypeDescriptor.singleOptionalValue(Mono.class),
ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
source -> (Mono<?>) source,
source -> source
Mono::from
);
registerReactiveType(ReactiveTypeDescriptor.multiValue(Flux.class),
registerReactiveType(ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
source -> (Flux<?>) source,
source -> source);
Flux::from);
registerReactiveType(ReactiveTypeDescriptor.multiValue(Publisher.class),
source -> Flux.from((Publisher<?>) source),
registerReactiveType(ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty),
source -> (Publisher<?>) source,
source -> source);
registerReactiveType(
ReactiveTypeDescriptor.singleOptionalValue(CompletableFuture.class),
ReactiveTypeDescriptor.singleOptionalValue(CompletableFuture.class, () -> {
CompletableFuture<?> empty = new CompletableFuture<>();
empty.complete(null);
return empty;
}),
source -> Mono.fromFuture((CompletableFuture<?>) source),
source -> Mono.from(source).toFuture()
);
@@ -98,17 +104,13 @@ public class ReactiveAdapterRegistry {
/**
* Register a reactive type along with functions to adapt to and from a
* Reactive Streams {@link Publisher}. The functions can assume their
* Reactive Streams {@link Publisher}. The functions can assume their
* input is never be {@code null} nor {@link Optional}.
*/
public void registerReactiveType(ReactiveTypeDescriptor descriptor,
Function<Object, Publisher<?>> toAdapter, Function<Publisher<?>, Object> fromAdapter) {
ReactiveAdapter adapter = (descriptor.isMultiValue() ?
new FluxReactiveAdapter(toAdapter, fromAdapter, descriptor) :
new MonoReactiveAdapter(toAdapter, fromAdapter, descriptor));
this.adapters.add(adapter);
this.adapters.add(new ReactorAdapter(descriptor, toAdapter, fromAdapter));
}
/**
@@ -123,7 +125,7 @@ public class ReactiveAdapterRegistry {
* "source" object is not {@code null} its actual type is used instead.
*/
public ReactiveAdapter getAdapterFrom(Class<?> reactiveType, Object source) {
source = unwrapOptional(source);
source = (source instanceof Optional ? ((Optional<?>) source).orElse(null) : source);
Class<?> clazz = (source != null ? source.getClass() : reactiveType);
return getAdapter(type -> type.isAssignableFrom(clazz));
}
@@ -142,132 +144,23 @@ public class ReactiveAdapterRegistry {
.orElse(null);
}
private static Object unwrapOptional(Object value) {
return (value instanceof Optional ? ((Optional<?>) value).orElse(null) : value);
}
@SuppressWarnings("unchecked")
private static class MonoReactiveAdapter implements ReactiveAdapter {
private final Function<Object, Publisher<?>> toAdapter;
private final Function<Publisher<?>, Object> fromAdapter;
private final ReactiveTypeDescriptor descriptor;
MonoReactiveAdapter(Function<Object, Publisher<?>> to, Function<Publisher<?>, Object> from,
ReactiveTypeDescriptor descriptor) {
this.toAdapter = to;
this.fromAdapter = from;
this.descriptor = descriptor;
}
@Override
public ReactiveTypeDescriptor getDescriptor() {
return this.descriptor;
}
@Override
public <T> Mono<T> toMono(Object source) {
source = unwrapOptional(source);
if (source == null) {
return Mono.empty();
}
return (Mono<T>) Mono.from(this.toAdapter.apply(source));
}
@Override
public <T> Flux<T> toFlux(Object source) {
source = unwrapOptional(source);
if (source == null) {
return Flux.empty();
}
return (Flux<T>) toMono(source).flux();
}
@Override
public <T> Publisher<T> toPublisher(Object source) {
return toMono(source);
}
@Override
public Object fromPublisher(Publisher<?> source) {
return (source != null ? this.fromAdapter.apply(source) : null);
}
}
@SuppressWarnings("unchecked")
private static class FluxReactiveAdapter implements ReactiveAdapter {
private final Function<Object, Publisher<?>> toAdapter;
private final Function<Publisher<?>, Object> fromAdapter;
private final ReactiveTypeDescriptor descriptor;
FluxReactiveAdapter(Function<Object, Publisher<?>> to, Function<Publisher<?>, Object> from,
ReactiveTypeDescriptor descriptor) {
this.descriptor = descriptor;
this.toAdapter = to;
this.fromAdapter = from;
}
@Override
public ReactiveTypeDescriptor getDescriptor() {
return this.descriptor;
}
@Override
public <T> Mono<T> toMono(Object source) {
source = unwrapOptional(source);
if (source == null) {
return Mono.empty();
}
return (Mono<T>) toFlux(source).next();
}
@Override
public <T> Flux<T> toFlux(Object source) {
source = unwrapOptional(source);
if (source == null) {
return Flux.empty();
}
return (Flux<T>) Flux.from(this.toAdapter.apply(source));
}
@Override
public <T> Publisher<T> toPublisher(Object source) {
return toFlux(source);
}
@Override
public Object fromPublisher(Publisher<?> source) {
return (source != null ? this.fromAdapter.apply(source) : null);
}
}
private static class RxJava1Registrar {
public void registerAdapters(ReactiveAdapterRegistry registry) {
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(rx.Observable.class),
source -> Flux.from(RxReactiveStreams.toPublisher((rx.Observable<?>) source)),
ReactiveTypeDescriptor.multiValue(rx.Observable.class, rx.Observable::empty),
source -> RxReactiveStreams.toPublisher((rx.Observable<?>) source),
RxReactiveStreams::toObservable
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleRequiredValue(rx.Single.class),
source -> Mono.from(RxReactiveStreams.toPublisher((rx.Single<?>) source)),
source -> RxReactiveStreams.toPublisher((rx.Single<?>) source),
RxReactiveStreams::toSingle
);
registry.registerReactiveType(
ReactiveTypeDescriptor.noValue(rx.Completable.class),
source -> Mono.from(RxReactiveStreams.toPublisher((rx.Completable) source)),
ReactiveTypeDescriptor.noValue(rx.Completable.class, Completable::complete),
source -> RxReactiveStreams.toPublisher((rx.Completable) source),
RxReactiveStreams::toCompletable
);
}
@@ -277,31 +170,52 @@ public class ReactiveAdapterRegistry {
public void registerAdapters(ReactiveAdapterRegistry registry) {
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(Flowable.class),
source -> Flux.from((Flowable<?>) source),
ReactiveTypeDescriptor.multiValue(Flowable.class, Flowable::empty),
source -> (Flowable<?>) source,
source-> Flowable.fromPublisher(source)
);
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class),
source -> Flux.from(((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER)),
ReactiveTypeDescriptor.multiValue(Observable.class, Observable::empty),
source -> ((Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
source -> Flowable.fromPublisher(source).toObservable()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
source -> Mono.from(((io.reactivex.Single<?>) source).toFlowable()),
source -> ((io.reactivex.Single<?>) source).toFlowable(),
source -> Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleOptionalValue(Maybe.class),
source -> Mono.from(((Maybe<?>) source).toFlowable()),
ReactiveTypeDescriptor.singleOptionalValue(Maybe.class, Maybe::empty),
source -> ((Maybe<?>) source).toFlowable(),
source -> Flowable.fromPublisher(source).toObservable().singleElement()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class),
source -> Mono.from(((io.reactivex.Completable) source).toFlowable()),
ReactiveTypeDescriptor.noValue(Completable.class, Completable::complete),
source -> ((Completable) source).toFlowable(),
source -> Flowable.fromPublisher(source).toObservable().ignoreElements()
);
}
}
/**
* Extension of ReactiveAdapter that wraps adapted (raw) Publisher's as
* {@link Flux} or {@link Mono} depending on the underlying reactive type's
* stream semantics.
*/
private static class ReactorAdapter extends ReactiveAdapter {
public ReactorAdapter(ReactiveTypeDescriptor descriptor,
Function<Object, Publisher<?>> toPublisherFunction,
Function<Publisher<?>, Object> fromPublisherFunction) {
super(descriptor, toPublisherFunction, fromPublisherFunction);
}
@Override
public <T> Publisher<T> toPublisher(Object source) {
Publisher<T> publisher = super.toPublisher(source);
return (getDescriptor().isMultiValue() ? Flux.from(publisher) : Mono.from(publisher));
}
}
}

42
spring-core/src/main/java/org/springframework/core/ReactiveTypeDescriptor.java
View File

@@ -15,6 +15,8 @@
*/
package org.springframework.core;
import java.util.function.Supplier;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -31,6 +33,8 @@ public class ReactiveTypeDescriptor {
private final Class<?> reactiveType;
private final Supplier<?> emptyValueSupplier;
private final boolean multiValue;
private final boolean supportsEmpty;
@@ -41,11 +45,13 @@ public class ReactiveTypeDescriptor {
/**
* Private constructor. See static factory methods.
*/
private ReactiveTypeDescriptor(Class<?> reactiveType, boolean multiValue,
boolean canBeEmpty, boolean noValue) {
private ReactiveTypeDescriptor(Class<?> reactiveType, Supplier<?> emptySupplier,
boolean multiValue, boolean canBeEmpty, boolean noValue) {
Assert.notNull(reactiveType, "'reactiveType' must not be null");
Assert.isTrue(!canBeEmpty || emptySupplier != null, "Empty value supplier is required.");
this.reactiveType = reactiveType;
this.emptyValueSupplier = emptySupplier;
this.multiValue = multiValue;
this.supportsEmpty = canBeEmpty;
this.noValue = noValue;
@@ -59,6 +65,15 @@ public class ReactiveTypeDescriptor {
return this.reactiveType;
}
/**
* Return an empty-value instance for the underlying reactive or async type.
* Use of this type implies {@link #supportsEmpty()} is true.
*/
public Object getEmptyValue() {
Assert.isTrue(supportsEmpty(), "Empty values not supported.");
return this.emptyValueSupplier.get();
}
/**
* Return {@code true} if the reactive type can produce more than 1 value
* can be produced and is therefore a good fit to adapt to {@link Flux}.
@@ -104,30 +119,37 @@ public class ReactiveTypeDescriptor {
/**
* Descriptor for a reactive type that can produce 0..N values.
* @param type the reactive type
* @param emptySupplier a supplier of an empty-value instance of the reactive type
*/
public static ReactiveTypeDescriptor multiValue(Class<?> reactiveType) {
return new ReactiveTypeDescriptor(reactiveType, true, true, false);
public static ReactiveTypeDescriptor multiValue(Class<?> type, Supplier<?> emptySupplier) {
return new ReactiveTypeDescriptor(type, emptySupplier, true, true, false);
}
/**
* Descriptor for a reactive type that can produce 0..1 values.
* @param type the reactive type
* @param emptySupplier a supplier of an empty-value instance of the reactive type
*/
public static ReactiveTypeDescriptor singleOptionalValue(Class<?> reactiveType) {
return new ReactiveTypeDescriptor(reactiveType, false, true, false);
public static ReactiveTypeDescriptor singleOptionalValue(Class<?> type, Supplier<?> emptySupplier) {
return new ReactiveTypeDescriptor(type, emptySupplier, false, true, false);
}
/**
* Descriptor for a reactive type that must produce 1 value to complete.
* @param type the reactive type
*/
public static ReactiveTypeDescriptor singleRequiredValue(Class<?> reactiveType) {
return new ReactiveTypeDescriptor(reactiveType, false, false, false);
public static ReactiveTypeDescriptor singleRequiredValue(Class<?> type) {
return new ReactiveTypeDescriptor(type, null, false, false, false);
}
/**
* Descriptor for a reactive type that does not produce any values.
* @param type the reactive type
* @param emptySupplier a supplier of an empty-value instance of the reactive type
*/
public static ReactiveTypeDescriptor noValue(Class<?> reactiveType) {
return new ReactiveTypeDescriptor(reactiveType, false, true, true);
public static ReactiveTypeDescriptor noValue(Class<?> type, Supplier<?> emptySupplier) {
return new ReactiveTypeDescriptor(type, emptySupplier, false, true, true);
}
}