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

Add RxJava1 Reactive Streams adapters check

Browse Source
This commit is contained in:
Rossen Stoyanchev
2017-04-04 12:25:58 -04:00
parent d9221fb87d
commit ae2306326e
1 changed files with 8 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -28,6 +28,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.RxReactiveStreams;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import static org.springframework.core.ReactiveTypeDescriptor.*;
@@ -52,7 +53,7 @@ public class ReactiveAdapterRegistry {
private static final boolean rxJava1Present =
ClassUtils.isPresent("rx.Observable", ReactiveAdapterRegistry.class.getClassLoader());
private static final boolean rxJava1Adapter =
private static final boolean rxReactiveStreamsPresent =
ClassUtils.isPresent("rx.RxReactiveStreams", ReactiveAdapterRegistry.class.getClassLoader());
private static final boolean rxJava2Present =
@@ -69,7 +70,7 @@ public class ReactiveAdapterRegistry {
if (reactorPresent) {
new ReactorRegistrar().registerAdapters(this);
}
if (rxJava1Present && rxJava1Adapter) {
if (rxJava1Present && rxReactiveStreamsPresent) {
new RxJava1Registrar().registerAdapters(this);
}
if (rxJava2Present) {
@@ -120,13 +121,17 @@ public class ReactiveAdapterRegistry {
* (i.e. to adapt from; may be {@code null} if the reactive type is specified)
*/
public ReactiveAdapter getAdapter(Class<?> reactiveType, Object source) {
Object sourceToUse = (source instanceof Optional ? ((Optional<?>) source).orElse(null) : source);
Class<?> clazz = (sourceToUse != null ? sourceToUse.getClass() : reactiveType);
if (clazz == null) {
return null;
}
Assert.isTrue(!rxJava1Present || rxReactiveStreamsPresent || !clazz.getName().startsWith("rx."),
"For RxJava 1.x adapter support please add " +
"\"io.reactivex:rxjava-reactive-streams\": " + clazz.getName());
return this.adapters.stream()
.filter(adapter -> adapter.getReactiveType() == clazz)
.findFirst()