diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java index fd2a9c76d4..15f98671a0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java @@ -16,8 +16,9 @@ package org.springframework.integration.channel; +import java.time.Duration; + import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -27,6 +28,7 @@ import org.springframework.messaging.SubscribableChannel; import reactor.core.publisher.EmitterProcessor; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; /** @@ -69,34 +71,12 @@ public final class MessageChannelReactiveUtils { }); } + @SuppressWarnings("unchecked") private static Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { - return new PollableChannelPublisherAdapter<>(inputChannel); - } - - private static final class PollableChannelPublisherAdapter implements Publisher> { - - private final PollableChannel channel; - - PollableChannelPublisherAdapter(final PollableChannel channel) { - this.channel = channel; - } - - @Override - @SuppressWarnings("unchecked") - public void subscribe(Subscriber> subscriber) { - Flux - .>create(sink -> - sink.onRequest(n -> { - Message m; - while (!sink.isCancelled() && n-- > 0 - && (m = this.channel.receive()) != null) { // NOSONAR - sink.next((Message) m); - } - })) - .subscribeOn(Schedulers.elastic()) - .subscribe(subscriber); - } - + return Mono.fromCallable(() -> (Message) inputChannel.receive()) + .subscribeOn(Schedulers.boundedElastic()) + .repeatWhenEmpty(it -> it.delayElements(Duration.ofMillis(10))) + .repeat(); } }