diff --git a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java index a959f5e2ad..fb69f7aad4 100644 --- a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java +++ b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java @@ -19,15 +19,15 @@ package org.springframework.core; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.List; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; import java.util.concurrent.Flow; import java.util.function.Function; +import org.reactivestreams.FlowAdapters; import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; import reactor.adapter.JdkFlowAdapter; import reactor.blockhound.BlockHound; import reactor.blockhound.integration.BlockHoundIntegration; @@ -114,7 +114,7 @@ public class ReactiveAdapterRegistry { // Simple Flow.Publisher bridge if Reactor is not present if (!reactorPresent) { - new FlowBridgeRegistrar().registerAdapter(this); + new FlowAdaptersRegistrar().registerAdapters(this); } } @@ -349,14 +349,19 @@ public class ReactiveAdapterRegistry { if (Flow.Publisher.class.isAssignableFrom(uniToPublisher.getReturnType())) { // Mutiny 2 based on Flow.Publisher - Method uniPublisher = ClassUtils.getMethod(io.smallrye.mutiny.groups.UniCreate.class, "publisher", Flow.Publisher.class); - Method multiPublisher = ClassUtils.getMethod(io.smallrye.mutiny.groups.MultiCreate.class, "publisher", Flow.Publisher.class); + Method uniPublisher = ClassUtils.getMethod( + io.smallrye.mutiny.groups.UniCreate.class, "publisher", Flow.Publisher.class); + Method multiPublisher = ClassUtils.getMethod( + io.smallrye.mutiny.groups.MultiCreate.class, "publisher", Flow.Publisher.class); registry.registerReactiveType(uniDesc, - uni -> new PublisherToRS<>((Flow.Publisher) ReflectionUtils.invokeMethod(uniToPublisher, ((io.smallrye.mutiny.Uni) uni).convert())), - publisher -> ReflectionUtils.invokeMethod(uniPublisher, io.smallrye.mutiny.Uni.createFrom(), new PublisherToFlow<>(publisher))); + uni -> FlowAdapters.toPublisher((Flow.Publisher) Objects.requireNonNull( + ReflectionUtils.invokeMethod(uniToPublisher, ((io.smallrye.mutiny.Uni) uni).convert()))), + publisher -> ReflectionUtils.invokeMethod(uniPublisher, io.smallrye.mutiny.Uni.createFrom(), + FlowAdapters.toFlowPublisher(publisher))); registry.registerReactiveType(multiDesc, - multi -> new PublisherToRS<>((Flow.Publisher) multi), - publisher -> ReflectionUtils.invokeMethod(multiPublisher, io.smallrye.mutiny.Multi.createFrom(), new PublisherToFlow<>(publisher))); + multi -> FlowAdapters.toPublisher((Flow.Publisher) multi), + publisher -> ReflectionUtils.invokeMethod(multiPublisher, io.smallrye.mutiny.Multi.createFrom(), + FlowAdapters.toFlowPublisher(publisher))); } else { // Mutiny 1 based on Reactive Streams @@ -371,19 +376,7 @@ public class ReactiveAdapterRegistry { } - private static class FlowBridgeRegistrar { - - @SuppressWarnings("unchecked") - void registerAdapter(ReactiveAdapterRegistry registry) { - registry.registerReactiveType( - ReactiveTypeDescriptor.multiValue(Flow.Publisher.class, () -> PublisherToRS.EMPTY_FLOW), - source -> new PublisherToRS<>((Flow.Publisher) source), - source -> new PublisherToFlow<>((Publisher) source)); - } - } - - - private static class PublisherToFlow implements Flow.Publisher { + private static class FlowAdaptersRegistrar { private static final Flow.Subscription EMPTY_SUBSCRIPTION = new Flow.Subscription() { @Override @@ -394,136 +387,17 @@ public class ReactiveAdapterRegistry { } }; - @Nullable - private final Publisher publisher; - - public PublisherToFlow(@Nullable Publisher publisher) { - this.publisher = publisher; - } - - @Override - public void subscribe(Flow.Subscriber subscriber) { - if (this.publisher != null) { - this.publisher.subscribe(new SubscriberToFlow<>(subscriber)); - } - else { - subscriber.onSubscribe(EMPTY_SUBSCRIPTION); - subscriber.onComplete(); - } - } - } - - - private static class PublisherToRS implements Publisher { - - private static final Flow.Publisher EMPTY_FLOW = new PublisherToFlow<>(null); - - private final Flow.Publisher publisher; + private static final Flow.Publisher EMPTY_PUBLISHER = subscriber -> { + subscriber.onSubscribe(EMPTY_SUBSCRIPTION); + subscriber.onComplete(); + }; @SuppressWarnings("unchecked") - public PublisherToRS(@Nullable Flow.Publisher publisher) { - this.publisher = (publisher != null ? publisher : (Flow.Publisher) EMPTY_FLOW); - } - - @Override - public void subscribe(Subscriber subscriber) { - this.publisher.subscribe(new SubscriberToRS<>(subscriber)); - } - } - - - private static class SubscriberToFlow implements Subscriber, Flow.Subscription { - - private final Flow.Subscriber subscriber; - - @Nullable - private Subscription subscription; - - public SubscriberToFlow(Flow.Subscriber subscriber) { - this.subscriber = subscriber; - } - - @Override - public void onSubscribe(Subscription subscription) { - this.subscription = subscription; - this.subscriber.onSubscribe(this); - } - - @Override - public void onNext(T o) { - this.subscriber.onNext(o); - } - - @Override - public void onError(Throwable t) { - this.subscriber.onError(t); - } - - @Override - public void onComplete() { - this.subscriber.onComplete(); - } - - @Override - public void request(long n) { - if (this.subscription != null) { - this.subscription.request(n); - } - } - - @Override - public void cancel() { - if (this.subscription != null) { - this.subscription.cancel(); - } - } - } - - - private static class SubscriberToRS implements Flow.Subscriber, Subscription { - - private final Subscriber subscriber; - - @Nullable - private Flow.Subscription subscription; - - public SubscriberToRS(Subscriber subscriber) { - this.subscriber = subscriber; - } - - @Override - public void onSubscribe(Flow.Subscription subscription) { - this.subscription = subscription; - this.subscriber.onSubscribe(this); - } - - @Override - public void onNext(T o) { - this.subscriber.onNext(o); - } - - @Override - public void onError(Throwable throwable) { - this.subscriber.onError(throwable); - } - - @Override - public void onComplete() { - this.subscriber.onComplete(); - } - - @Override - public void request(long n) { - if (this.subscription != null) { - this.subscription.request(n); - } - } - - @Override - public void cancel() { - if (this.subscription != null) { - this.subscription.cancel(); - } + void registerAdapters(ReactiveAdapterRegistry registry) { + registry.registerReactiveType( + ReactiveTypeDescriptor.multiValue(Flow.Publisher.class, () -> EMPTY_PUBLISHER), + source -> FlowAdapters.toPublisher((Flow.Publisher) source), + source -> FlowAdapters.toFlowPublisher((Publisher) source)); } }