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

Polish ReactiveAdapterRegisry

Browse Source
This commit is contained in:
Rossen Stoyanchev
2018-05-18 09:29:39 -04:00
parent e3e975d7f9
commit 1dc8201df1
4 changed files with 79 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -24,11 +24,10 @@ import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Adapt a Reactive Streams {@link Publisher} to and from an async/reactive type
* such as {@code CompletableFuture}, an RxJava {@code Observable}, etc.
* Adapter for a Reactive Streams {@link Publisher} to and from an async/reactive
* type such as {@code CompletableFuture}, RxJava {@code Observable}, and others.
*
* <p>Use the {@link ReactiveAdapterRegistry} to register reactive types and
* obtain adapters from.
* <p>An adapter is typically obtained via {@link ReactiveAdapterRegistry}.
*
* @author Rossen Stoyanchev
* @since 5.0
@@ -71,28 +70,28 @@ public class ReactiveAdapter {
}
/**
* A shortcut for {@code getDescriptor().getReactiveType()}.
* Shortcut for {@code getDescriptor().getReactiveType()}.
*/
public Class<?> getReactiveType() {
return getDescriptor().getReactiveType();
}
/**
* A shortcut for {@code getDescriptor().isMultiValue()}.
* Shortcut for {@code getDescriptor().isMultiValue()}.
*/
public boolean isMultiValue() {
return getDescriptor().isMultiValue();
}
/**
* A shortcut for {@code getDescriptor().supportsEmpty()}.
* Shortcut for {@code getDescriptor().supportsEmpty()}.
*/
public boolean supportsEmpty() {
return getDescriptor().supportsEmpty();
}
/**
* A shortcut for {@code getDescriptor().isNoValue()}.
* Shortcut for {@code getDescriptor().isNoValue()}.
*/
public boolean isNoValue() {
return getDescriptor().isNoValue();
@@ -100,8 +99,9 @@ public class ReactiveAdapter {
/**
* Adapt the given instance to a Reactive Streams Publisher.
* @param source the source object to adapt from
* Adapt the given instance to a Reactive Streams {@code Publisher}.
* @param source the source object to adapt from; if the given object is
* {@code null}, {@link ReactiveTypeDescriptor#getEmptyValue()} is used.
* @return the Publisher representing the adaptation
*/
@SuppressWarnings("unchecked")

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -34,13 +34,8 @@ import org.springframework.lang.Nullable;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import static org.springframework.core.ReactiveTypeDescriptor.multiValue;
import static org.springframework.core.ReactiveTypeDescriptor.noValue;
import static org.springframework.core.ReactiveTypeDescriptor.singleOptionalValue;
import static org.springframework.core.ReactiveTypeDescriptor.singleRequiredValue;
/**
* A registry of adapters to adapt a Reactive Streams {@link Publisher} to/from
* A registry of adapters to adapt Reactive Streams {@link Publisher} to/from
* various async/reactive types such as {@code CompletableFuture}, RxJava
* {@code Observable}, and others.
*
@@ -64,6 +59,7 @@ public class ReactiveAdapterRegistry {
/**
* Create a registry and auto-register default adapters.
* @see #getSharedInstance()
*/
public ReactiveAdapterRegistry() {
@@ -168,49 +164,54 @@ public class ReactiveAdapterRegistry {
/**
* Return a shared default {@code ReactiveAdapterRegistry} instance, lazily
* building it once needed.
*
* <p><b>NOTE:</b> We highly recommend passing a long-lived, pre-configured
* {@code ReactiveAdapterRegistry} instance for customization purposes.
* This accessor is only meant as a fallback for code paths that want to
* fall back on a default instance if one isn't provided.
*
* @return the shared {@code ReactiveAdapterRegistry} instance (never {@code null})
* @since 5.0.2
*/
public static ReactiveAdapterRegistry getSharedInstance() {
ReactiveAdapterRegistry ar = sharedInstance;
if (ar == null) {
ReactiveAdapterRegistry registry = sharedInstance;
if (registry == null) {
synchronized (ReactiveAdapterRegistry.class) {
ar = sharedInstance;
if (ar == null) {
ar = new ReactiveAdapterRegistry();
sharedInstance = ar;
registry = sharedInstance;
if (registry == null) {
registry = new ReactiveAdapterRegistry();
sharedInstance = registry;
}
}
}
return ar;
return registry;
}
private static class ReactorRegistrar {
void registerAdapters(ReactiveAdapterRegistry registry) {
// Flux and Mono ahead of Publisher...
// Register Flux and Mono before Publisher...
registry.registerReactiveType(
singleOptionalValue(Mono.class, Mono::empty),
ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
source -> (Mono<?>) source,
Mono::from
);
registry.registerReactiveType(multiValue(Flux.class, Flux::empty),
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
source -> (Flux<?>) source,
Flux::from);
registry.registerReactiveType(multiValue(Publisher.class, Flux::empty),
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty),
source -> (Publisher<?>) source,
source -> source);
registry.registerReactiveType(
singleOptionalValue(CompletableFuture.class, () -> {
ReactiveTypeDescriptor.singleOptionalValue(CompletableFuture.class, () -> {
CompletableFuture<?> empty = new CompletableFuture<>();
empty.complete(null);
return empty;
@@ -226,17 +227,17 @@ public class ReactiveAdapterRegistry {
void registerAdapters(ReactiveAdapterRegistry registry) {
registry.registerReactiveType(
multiValue(rx.Observable.class, rx.Observable::empty),
ReactiveTypeDescriptor.multiValue(rx.Observable.class, rx.Observable::empty),
source -> RxReactiveStreams.toPublisher((rx.Observable<?>) source),
RxReactiveStreams::toObservable
);
registry.registerReactiveType(
singleRequiredValue(rx.Single.class),
ReactiveTypeDescriptor.singleRequiredValue(rx.Single.class),
source -> RxReactiveStreams.toPublisher((rx.Single<?>) source),
RxReactiveStreams::toSingle
);
registry.registerReactiveType(
noValue(rx.Completable.class, rx.Completable::complete),
ReactiveTypeDescriptor.noValue(rx.Completable.class, rx.Completable::complete),
source -> RxReactiveStreams.toPublisher((rx.Completable) source),
RxReactiveStreams::toCompletable
);
@@ -248,27 +249,27 @@ public class ReactiveAdapterRegistry {
void registerAdapters(ReactiveAdapterRegistry registry) {
registry.registerReactiveType(
multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
source -> (io.reactivex.Flowable<?>) source,
Flowable::fromPublisher
);
registry.registerReactiveType(
multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
source -> ((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable()
);
registry.registerReactiveType(
singleRequiredValue(io.reactivex.Single.class),
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
source -> ((io.reactivex.Single<?>) source).toFlowable(),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
);
registry.registerReactiveType(
singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
source -> ((io.reactivex.Maybe<?>) source).toFlowable(),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement()
);
registry.registerReactiveType(
noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
source -> ((io.reactivex.Completable) source).toFlowable(),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable().ignoreElements()
);
@@ -278,15 +279,15 @@ public class ReactiveAdapterRegistry {
private static class ReactorJdkFlowAdapterRegistrar {
// TODO: remove reflection when build requires JDK 9+
void registerAdapter(ReactiveAdapterRegistry registry) throws Exception {
// TODO: remove reflection when build requires JDK 9+
Class<?> type = ClassUtils.forName("java.util.concurrent.Flow.Publisher", getClass().getClassLoader());
Method toFluxMethod = getMethod("flowPublisherToFlux", type);
Method toFlowMethod = getMethod("publisherToFlowPublisher", Publisher.class);
Object emptyFlow = ReflectionUtils.invokeMethod(toFlowMethod, null, Flux.empty());
registry.registerReactiveType(
multiValue(type, () -> emptyFlow),
ReactiveTypeDescriptor.multiValue(type, () -> emptyFlow),
source -> (Publisher<?>) ReflectionUtils.invokeMethod(toFluxMethod, null, source),
publisher -> ReflectionUtils.invokeMethod(toFlowMethod, null, publisher)
);
@@ -299,9 +300,10 @@ public class ReactiveAdapterRegistry {
/**
* Extension of ReactiveAdapter that wraps adapted (raw) Publisher's as
* {@link Flux} or {@link Mono} depending on the underlying reactive type's
* stream semantics.
* ReactiveAdapter variant that wraps adapted Publishers as {@link Flux} or
* {@link Mono} depending on {@link ReactiveTypeDescriptor#isMultiValue()}.
* This is important in places where only the stream and stream element type
* information is available like encoders and decoders.
*/
private static class ReactorAdapter extends ReactiveAdapter {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -22,8 +22,8 @@ import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Descriptor for a reactive type with information about its stream semantics, i.e.
* how many values it can produce.
* Describes the semantics of a reactive type including boolean checks for
* {@link #isMultiValue()}, {@link #supportsEmpty()}, and {@link #isNoValue()}.
*
* @author Rossen Stoyanchev
* @since 5.0
@@ -55,21 +55,12 @@ public class ReactiveTypeDescriptor {
/**
* Return the reactive type the descriptor was created for.
* Return the reactive type for this descriptor.
*/
public Class<?> getReactiveType() {
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.state(this.emptyValueSupplier != null, "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 {@code Flux}.
@@ -95,6 +86,15 @@ public class ReactiveTypeDescriptor {
return this.noValue;
}
/**
* 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.state(this.emptyValueSupplier != null, "Empty values not supported");
return this.emptyValueSupplier.get();
}
@Override
public boolean equals(@Nullable Object other) {