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

Fix AbstractMProducingHandler for ReactiveAdapter

Browse Source 
Turns out there is a reactive adapter for `CompletableFuture`
which is not an intention of the further logic
This commit is contained in:
Artem Bilan
2022-11-03 13:20:45 -04:00
parent 5f1c0c17de
commit 53d044814f
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -310,11 +310,16 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
} }
if (this.async) { if (this.async) {
ReactiveAdapter reactiveAdapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply); boolean isFutureReply =
if (reply instanceof org.springframework.util.concurrent.ListenableFuture<?> reply instanceof org.springframework.util.concurrent.ListenableFuture<?> ||
|| reply instanceof CompletableFuture<?> reply instanceof CompletableFuture<?>;
|| reactiveAdapter != null) {
ReactiveAdapter reactiveAdapter = null;
if (!isFutureReply) {
reactiveAdapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply);
}
if (isFutureReply || reactiveAdapter != null) {
if (replyChannel instanceof ReactiveStreamsSubscribableChannel reactiveStreamsSubscribableChannel) { if (replyChannel instanceof ReactiveStreamsSubscribableChannel reactiveStreamsSubscribableChannel) {
Publisher<?> reactiveReply = toPublisherReply(reply, reactiveAdapter); Publisher<?> reactiveReply = toPublisherReply(reply, reactiveAdapter);
reactiveStreamsSubscribableChannel reactiveStreamsSubscribableChannel