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

INT-4208: Add WebFlux-Based HttpMessageHandler

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4208

Since `AsyncRestTemplate` is deprecated in Spring 5.0, it doesn't make
sense to promote that feature via our new `AsyncHttpRequestExecutingMessageHandler`
component

* Rework (and rename) `AsyncHttpRequestExecutingMessageHandler` to
`ReactiveHttpRequestExecutingMessageHandler` and make it based on the
WebFlux `WebClient`
* Fix Java DSL (`ReactiveHttpMessageHandlerSpec`) and all tests according a new
logic in the `ReactiveHttpRequestExecutingMessageHandler`
* Fix XML namespace support to use new `ReactiveHttpRequestExecutingMessageHandler`
and don't expose unused options like `converters` and `request-factory`
* Fix `What's New` and `http.adoc`
* To remain with `async` mode for the `ReactiveHttpRequestExecutingMessageHandler`
behavior support fix `AbstractMessageProducingHandler` to adapt reply `Mono`
to the `SettableListenableFuture`
* Introduce new `reactive` behavior for the `AbstractMessageProducingHandler`
when `outputChannel` is `ReactiveSubscribableChannel` and perform
`.subscribeTo(Publisher)` for `Publisher` reply
* Upgrade to Spring AMQP `2.0 M3`
* Downgrade to Spring Security `4.2.2`

Address PR comments

Add async error handling to the one-way case

Minor polishing and checkstyle fix
This commit is contained in:
Artem Bilan
2017-04-04 15:27:29 -04:00
committed by Gary Russell
parent d954e6a66f
commit 19c5402079
25 changed files with 939 additions and 813 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,7 +21,10 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.reactivestreams.Publisher;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.ReactiveSubscribableChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.routingslip.RoutingSlipRouteStrategy;
@@ -37,6 +40,10 @@ import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.util.concurrent.SettableListenableFuture;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}.
@@ -44,6 +51,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
* @author David Liu
* @author Artem Bilan
* @author Gary Russell
*
* since 4.1
*/
public abstract class AbstractMessageProducingHandler extends AbstractMessageHandler
@@ -181,36 +189,56 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
}
}
if (this.async && reply instanceof ListenableFuture<?>) {
ListenableFuture<?> future = (ListenableFuture<?>) reply;
final Object theReplyChannel = replyChannel;
future.addCallback(new ListenableFutureCallback<Object>() {
if (this.async && (reply instanceof ListenableFuture<?> || reply instanceof Publisher<?>)) {
if (reply instanceof ListenableFuture<?> || !(getOutputChannel() instanceof ReactiveSubscribableChannel)) {
ListenableFuture<?> future;
if (reply instanceof ListenableFuture<?>) {
future = (ListenableFuture<?>) reply;
}
else {
SettableListenableFuture<Object> settableListenableFuture = new SettableListenableFuture<>();
@Override
public void onSuccess(Object result) {
Message<?> replyMessage = null;
try {
replyMessage = createOutputMessage(result, requestHeaders);
sendOutput(replyMessage, theReplyChannel, false);
}
catch (Exception e) {
Exception exceptionToLogAndSend = e;
if (!(e instanceof MessagingException)) {
exceptionToLogAndSend = new MessageHandlingException(requestMessage, e);
if (replyMessage != null) {
exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
}
Mono.from((Publisher<?>) reply)
.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
future = settableListenableFuture;
}
Object theReplyChannel = replyChannel;
future.addCallback(new ListenableFutureCallback<Object>() {
@Override
public void onSuccess(Object result) {
Message<?> replyMessage = null;
try {
replyMessage = createOutputMessage(result, requestHeaders);
sendOutput(replyMessage, theReplyChannel, false);
}
catch (Exception e) {
Exception exceptionToLogAndSend = e;
if (!(e instanceof MessagingException)) {
exceptionToLogAndSend = new MessageHandlingException(requestMessage, e);
if (replyMessage != null) {
exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
}
}
logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);
onFailure(exceptionToLogAndSend);
}
logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);
onFailure(exceptionToLogAndSend);
}
}
@Override
public void onFailure(Throwable ex) {
sendErrorMessage(requestMessage, ex);
}
});
@Override
public void onFailure(Throwable ex) {
sendErrorMessage(requestMessage, ex);
}
});
}
else {
((ReactiveSubscribableChannel) getOutputChannel())
.subscribeTo(Flux.from((Publisher<?>) reply)
.map(result -> createOutputMessage(result, requestHeaders)));
}
}
else {
sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false);