From e6e9f4516458d5e8d8bb525ce19935228b39975d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 2 Mar 2020 16:57:29 -0500 Subject: [PATCH] Use Mono.create instead of Mono.fromCallable It is better to poll message from `QueueChannel` when ever a reactive request happens. The `Mono.fromCallable()` operator does poll on subscription, not request and caches the value. However we may lose such a value in between. * Use `Mono.create()` with its `monoSink.onRequest()` callback in the `adaptPollableChannelToPublisher()` implementation to defer `inputChannel.receive()` until an on demand request downstream --- .../integration/channel/MessageChannelReactiveUtils.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 2b11d1e090..797dcc1ca0 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 @@ -73,7 +73,9 @@ public final class MessageChannelReactiveUtils { @SuppressWarnings("unchecked") private static Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { - return Mono.fromCallable(() -> (Message) inputChannel.receive(0)) + return Mono.>create(monoSink -> + monoSink.onRequest(value -> + monoSink.success((Message) inputChannel.receive(0)))) .subscribeOn(Schedulers.boundedElastic()) .repeatWhenEmpty(it -> it.delayElements(Duration.ofMillis(100))) // NOSONAR - magic .repeat();