Remove support for RxJava 2.x as well
Includes realignment of Reactor adapter registration. Closes gh-27443
This commit is contained in:
Juergen Hoeller
@@ -58,8 +58,6 @@ public class ReactiveAdapterRegistry {
|
||||
|
||||
private static final boolean reactorPresent;
|
||||
|
||||
private static final boolean rxjava2Present;
|
||||
|
||||
private static final boolean rxjava3Present;
|
||||
|
||||
private static final boolean kotlinCoroutinesPresent;
|
||||
@@ -69,7 +67,6 @@ public class ReactiveAdapterRegistry {
|
||||
static {
|
||||
ClassLoader classLoader = ReactiveAdapterRegistry.class.getClassLoader();
|
||||
reactorPresent = ClassUtils.isPresent("reactor.core.publisher.Flux", classLoader);
|
||||
rxjava2Present = ClassUtils.isPresent("io.reactivex.Flowable", classLoader);
|
||||
rxjava3Present = ClassUtils.isPresent("io.reactivex.rxjava3.core.Flowable", classLoader);
|
||||
kotlinCoroutinesPresent = ClassUtils.isPresent("kotlinx.coroutines.reactor.MonoKt", classLoader);
|
||||
mutinyPresent = ClassUtils.isPresent("io.smallrye.mutiny.Multi", classLoader);
|
||||
@@ -86,13 +83,9 @@ public class ReactiveAdapterRegistry {
|
||||
// Reactor
|
||||
if (reactorPresent) {
|
||||
new ReactorRegistrar().registerAdapters(this);
|
||||
new ReactorJdkFlowAdapterRegistrar().registerAdapter(this);
|
||||
}
|
||||
|
||||
// RxJava
|
||||
if (rxjava2Present) {
|
||||
new RxJava2Registrar().registerAdapters(this);
|
||||
}
|
||||
if (rxjava3Present) {
|
||||
new RxJava3Registrar().registerAdapters(this);
|
||||
}
|
||||
@@ -200,123 +193,6 @@ public class ReactiveAdapterRegistry {
|
||||
}
|
||||
|
||||
|
||||
private static class ReactorRegistrar {
|
||||
|
||||
void registerAdapters(ReactiveAdapterRegistry registry) {
|
||||
// Register Flux and Mono before Publisher...
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
|
||||
source -> (Mono<?>) source,
|
||||
Mono::from
|
||||
);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
|
||||
source -> (Flux<?>) source,
|
||||
Flux::from);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty),
|
||||
source -> (Publisher<?>) source,
|
||||
source -> source);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.nonDeferredAsyncValue(CompletionStage.class, EmptyCompletableFuture::new),
|
||||
source -> Mono.fromCompletionStage((CompletionStage<?>) source),
|
||||
source -> Mono.from(source).toFuture()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class ReactorJdkFlowAdapterRegistrar {
|
||||
|
||||
void registerAdapter(ReactiveAdapterRegistry registry) {
|
||||
Flow.Publisher<?> emptyFlow = JdkFlowAdapter.publisherToFlowPublisher(Flux.empty());
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(Flow.Publisher.class, () -> emptyFlow),
|
||||
source -> JdkFlowAdapter.flowPublisherToFlux((Flow.Publisher<?>) source),
|
||||
JdkFlowAdapter::publisherToFlowPublisher
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class RxJava2Registrar {
|
||||
|
||||
void registerAdapters(ReactiveAdapterRegistry registry) {
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
|
||||
source -> (io.reactivex.Flowable<?>) source,
|
||||
io.reactivex.Flowable::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
|
||||
source -> ((io.reactivex.Observable<?>) source).toFlowable(io.reactivex.BackpressureStrategy.BUFFER),
|
||||
io.reactivex.Observable::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
|
||||
source -> ((io.reactivex.Single<?>) source).toFlowable(),
|
||||
io.reactivex.Single::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
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(
|
||||
ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
|
||||
source -> ((io.reactivex.Completable) source).toFlowable(),
|
||||
io.reactivex.Completable::fromPublisher
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class RxJava3Registrar {
|
||||
|
||||
void registerAdapters(ReactiveAdapterRegistry registry) {
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(
|
||||
io.reactivex.rxjava3.core.Flowable.class,
|
||||
io.reactivex.rxjava3.core.Flowable::empty),
|
||||
source -> (io.reactivex.rxjava3.core.Flowable<?>) source,
|
||||
io.reactivex.rxjava3.core.Flowable::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(
|
||||
io.reactivex.rxjava3.core.Observable.class,
|
||||
io.reactivex.rxjava3.core.Observable::empty),
|
||||
source -> ((io.reactivex.rxjava3.core.Observable<?>) source).toFlowable(
|
||||
io.reactivex.rxjava3.core.BackpressureStrategy.BUFFER),
|
||||
io.reactivex.rxjava3.core.Observable::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.rxjava3.core.Single.class),
|
||||
source -> ((io.reactivex.rxjava3.core.Single<?>) source).toFlowable(),
|
||||
io.reactivex.rxjava3.core.Single::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleOptionalValue(
|
||||
io.reactivex.rxjava3.core.Maybe.class,
|
||||
io.reactivex.rxjava3.core.Maybe::empty),
|
||||
source -> ((io.reactivex.rxjava3.core.Maybe<?>) source).toFlowable(),
|
||||
io.reactivex.rxjava3.core.Maybe::fromPublisher
|
||||
);
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.noValue(
|
||||
io.reactivex.rxjava3.core.Completable.class,
|
||||
io.reactivex.rxjava3.core.Completable::complete),
|
||||
source -> ((io.reactivex.rxjava3.core.Completable) source).toFlowable(),
|
||||
io.reactivex.rxjava3.core.Completable::fromPublisher
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* ReactiveAdapter variant that wraps adapted Publishers as {@link Flux} or
|
||||
* {@link Mono} depending on {@link ReactiveTypeDescriptor#isMultiValue()}.
|
||||
@@ -340,6 +216,41 @@ public class ReactiveAdapterRegistry {
|
||||
}
|
||||
|
||||
|
||||
private static class ReactorRegistrar {
|
||||
|
||||
private static final Flow.Publisher<?> EMPTY_FLOW = JdkFlowAdapter.publisherToFlowPublisher(Flux.empty());
|
||||
|
||||
void registerAdapters(ReactiveAdapterRegistry registry) {
|
||||
// Register Flux and Mono before Publisher...
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty),
|
||||
source -> (Mono<?>) source,
|
||||
Mono::from);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty),
|
||||
source -> (Flux<?>) source,
|
||||
Flux::from);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty),
|
||||
source -> (Publisher<?>) source,
|
||||
source -> source);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.nonDeferredAsyncValue(CompletionStage.class, EmptyCompletableFuture::new),
|
||||
source -> Mono.fromCompletionStage((CompletionStage<?>) source),
|
||||
source -> Mono.from(source).toFuture());
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(Flow.Publisher.class, () -> EMPTY_FLOW),
|
||||
source -> JdkFlowAdapter.flowPublisherToFlux((Flow.Publisher<?>) source),
|
||||
JdkFlowAdapter::publisherToFlowPublisher);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class EmptyCompletableFuture<T> extends CompletableFuture<T> {
|
||||
|
||||
EmptyCompletableFuture() {
|
||||
@@ -348,6 +259,46 @@ public class ReactiveAdapterRegistry {
|
||||
}
|
||||
|
||||
|
||||
private static class RxJava3Registrar {
|
||||
|
||||
void registerAdapters(ReactiveAdapterRegistry registry) {
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(
|
||||
io.reactivex.rxjava3.core.Flowable.class,
|
||||
io.reactivex.rxjava3.core.Flowable::empty),
|
||||
source -> (io.reactivex.rxjava3.core.Flowable<?>) source,
|
||||
io.reactivex.rxjava3.core.Flowable::fromPublisher);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(
|
||||
io.reactivex.rxjava3.core.Observable.class,
|
||||
io.reactivex.rxjava3.core.Observable::empty),
|
||||
source -> ((io.reactivex.rxjava3.core.Observable<?>) source).toFlowable(
|
||||
io.reactivex.rxjava3.core.BackpressureStrategy.BUFFER),
|
||||
io.reactivex.rxjava3.core.Observable::fromPublisher);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.rxjava3.core.Single.class),
|
||||
source -> ((io.reactivex.rxjava3.core.Single<?>) source).toFlowable(),
|
||||
io.reactivex.rxjava3.core.Single::fromPublisher);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.singleOptionalValue(
|
||||
io.reactivex.rxjava3.core.Maybe.class,
|
||||
io.reactivex.rxjava3.core.Maybe::empty),
|
||||
source -> ((io.reactivex.rxjava3.core.Maybe<?>) source).toFlowable(),
|
||||
io.reactivex.rxjava3.core.Maybe::fromPublisher);
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.noValue(
|
||||
io.reactivex.rxjava3.core.Completable.class,
|
||||
io.reactivex.rxjava3.core.Completable::complete),
|
||||
source -> ((io.reactivex.rxjava3.core.Completable) source).toFlowable(),
|
||||
io.reactivex.rxjava3.core.Completable::fromPublisher);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class CoroutinesRegistrar {
|
||||
|
||||
@SuppressWarnings("KotlinInternalInJava")
|
||||
@@ -361,8 +312,7 @@ public class ReactiveAdapterRegistry {
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(kotlinx.coroutines.flow.Flow.class, kotlinx.coroutines.flow.FlowKt::emptyFlow),
|
||||
source -> kotlinx.coroutines.reactor.ReactorFlowKt.asFlux((kotlinx.coroutines.flow.Flow<?>) source),
|
||||
kotlinx.coroutines.reactive.ReactiveFlowKt::asFlow
|
||||
);
|
||||
kotlinx.coroutines.reactive.ReactiveFlowKt::asFlow);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -375,16 +325,14 @@ public class ReactiveAdapterRegistry {
|
||||
io.smallrye.mutiny.Uni.class,
|
||||
() -> io.smallrye.mutiny.Uni.createFrom().nothing()),
|
||||
uni -> ((io.smallrye.mutiny.Uni<?>) uni).convert().toPublisher(),
|
||||
publisher -> io.smallrye.mutiny.Uni.createFrom().publisher(publisher)
|
||||
);
|
||||
publisher -> io.smallrye.mutiny.Uni.createFrom().publisher(publisher));
|
||||
|
||||
registry.registerReactiveType(
|
||||
ReactiveTypeDescriptor.multiValue(
|
||||
io.smallrye.mutiny.Multi.class,
|
||||
() -> io.smallrye.mutiny.Multi.createFrom().empty()),
|
||||
multi -> (io.smallrye.mutiny.Multi<?>) multi,
|
||||
publisher -> io.smallrye.mutiny.Multi.createFrom().publisher(publisher)
|
||||
);
|
||||
publisher -> io.smallrye.mutiny.Multi.createFrom().publisher(publisher));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -402,7 +350,6 @@ public class ReactiveAdapterRegistry {
|
||||
|
||||
@Override
|
||||
public void applyTo(BlockHound.Builder builder) {
|
||||
|
||||
// Avoid hard references potentially anywhere in spring-core (no need for structural dependency)
|
||||
|
||||
builder.allowBlockingCallsInside(
|
||||
|
||||
Reference in New Issue
Block a user