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

Fix MessagingGateway for reactive reply type

Browse Source 
* Upgrade dependencies to be ready for release
This commit is contained in:
Artem Bilan
2020-06-23 14:26:11 -04:00
parent 3bb445e133
commit 9367d7f4de
5 changed files with 32 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -875,8 +875,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
@Override
public void subscribeTo(Publisher<? extends Message<?>> publisher) {
this.replyMono.switchIfEmpty(Mono.from(publisher));
this.replyMono.onComplete();
publisher.subscribe(this.replyMono);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -299,21 +299,26 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
}
private void doProduceOutput(Message<?> requestMessage, MessageHeaders requestHeaders, Object reply,
@Nullable Object replyChannel) {
@Nullable Object replyChannelArg) {
Object replyChannel = replyChannelArg;
if (replyChannel == null) {
replyChannel = getOutputChannel();
}
if (this.async && (reply instanceof ListenableFuture<?> || reply instanceof Publisher<?>)) {
MessageChannel messageChannel = getOutputChannel();
if (reply instanceof ListenableFuture<?> ||
!(messageChannel instanceof ReactiveStreamsSubscribableChannel)) {
asyncNonReactiveReply(requestMessage, reply, replyChannel);
}
else {
((ReactiveStreamsSubscribableChannel) messageChannel)
if (reply instanceof Publisher<?> &&
replyChannel instanceof ReactiveStreamsSubscribableChannel) {
((ReactiveStreamsSubscribableChannel) replyChannel)
.subscribeTo(
Flux.from((Publisher<?>) reply)
.doOnError((ex) -> sendErrorMessage(requestMessage, ex))
.map(result -> createOutputMessage(result, requestHeaders)));
}
else {
asyncNonReactiveReply(requestMessage, reply, replyChannel);
}
}
else {
sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false);