From ae2e1b0a8d83299c6d03642307023266b111657e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 27 Jan 2016 12:42:55 -0500 Subject: [PATCH] INT-3947: Support Async Message Handlers JIRA: https://jira.spring.io/browse/INT-3947 Async Message Handler If `asyncReplySupported` and the handler returns a `ListenableFuture`, defer the send to the reply channel until the future is satisfied. First candidate would be the `AmqpOutboundGateway` wired with an `AsyncRabbitTemplate`. Polishing - Use errorChannel Header for Exceptions Add more tests. Polishing - ListenableFuture Callback Exceptions Polishing - send error if error on async output Cover `onFailure()` from `onSuccess()` with the `errorChannel` --- .../AbstractMessageProducingHandler.java | 80 ++++- .../handler/AsyncHandlerTests.java | 276 ++++++++++++++++++ 2 files changed, 349 insertions(+), 7 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index aaa034e5eb..a5fe8157d6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -28,11 +28,14 @@ import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; /** * The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}. @@ -51,6 +54,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan private volatile String outputChannelName; + private volatile boolean asyncReplySupported; + /** * Set the timeout for sending reply Messages. * @param sendTimeout The send timeout. @@ -69,6 +74,18 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan this.outputChannelName = outputChannelName;//NOSONAR (inconsistent sync) } + /** + * Allow async replies. If the handler reply is a {@link ListenableFuture} send + * the output when it is satisfied rather than sending the future as the result. + * Only subclasses that support this feature should set it. + * @param asyncReplySupported true to allow. + * + * @since 4.3 + */ + protected void setAsyncReplySupported(boolean asyncReplySupported) { + this.asyncReplySupported = asyncReplySupported; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -112,8 +129,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return false; } - protected void produceOutput(Object reply, Message requestMessage) { - MessageHeaders requestHeaders = requestMessage.getHeaders(); + protected void produceOutput(Object reply, final Message requestMessage) { + final MessageHeaders requestHeaders = requestMessage.getHeaders(); Object replyChannel = null; if (getOutputChannel() == null) { @@ -150,8 +167,55 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } } - Message replyMessage = createOutputMessage(reply, requestHeaders); - sendOutput(replyMessage, replyChannel); + if (this.asyncReplySupported && reply instanceof ListenableFuture) { + ListenableFuture future = (ListenableFuture) reply; + final Object theReplyChannel = replyChannel; + future.addCallback(new ListenableFutureCallback() { + + @Override + public void onSuccess(Object result) { + try { + sendOutput(createOutputMessage(result, requestHeaders), theReplyChannel, false); + } + catch (Exception e) { + Exception exceptionToLogAndSend = e; + if (!(e instanceof MessagingException)) { + exceptionToLogAndSend = new MessageHandlingException(requestMessage, e); + } + logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend); + onFailure(exceptionToLogAndSend); + } + } + + @Override + public void onFailure(Throwable ex) { + Object errorChannel = requestHeaders.getErrorChannel(); + Throwable result = ex; + if (!(ex instanceof MessagingException)) { + result = new MessageHandlingException(requestMessage, ex); + } + if (errorChannel == null) { + logger.error("Async exception received and no 'errorChannel' header exists; cannot route " + + "exception to caller", result); + } + else { + try { + sendOutput(createOutputMessage(result, requestHeaders), errorChannel, true); + } + catch (Exception e) { + Exception exceptionToLog = e; + if (!(e instanceof MessagingException)) { + exceptionToLog = new MessageHandlingException(requestMessage, e); + } + logger.error("Failed to send async reply", exceptionToLog); + } + } + } + }); + } + else { + sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false); + } } private Object getOutputChannelFromRoutingSlip(Object reply, Message requestMessage, List routingSlip, @@ -216,10 +280,12 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan * null, and it must be an instance of either String or {@link MessageChannel}. * @param output the output object to send * @param replyChannel the 'replyChannel' value from the original request + * @param isError - this is an error, use the replyChannel argument (must not be null), not + * the configured output channel. */ - private void sendOutput(Object output, Object replyChannel) { + private void sendOutput(Object output, Object replyChannel, boolean isError) { MessageChannel outputChannel = getOutputChannel(); - if (outputChannel != null) { + if (!isError && outputChannel != null) { replyChannel = outputChannel; } if (replyChannel == null) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java new file mode 100644 index 0000000000..2bfd3f6d42 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java @@ -0,0 +1,276 @@ +/* + * Copyright 2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.junit.Before; +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.gateway.GatewayProxyFactoryBean; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.core.DestinationResolutionException; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.concurrent.SettableListenableFuture; + +/** + * @author Gary Russell + * @since 4.3 + * + */ +public class AsyncHandlerTests { + + private final QueueChannel output = new QueueChannel(); + + private AbstractReplyProducingMessageHandler handler; + + private volatile CountDownLatch latch; + + private volatile int whichTest; + + private volatile Exception failedCallbackException; + + private volatile String failedCallbackMessage; + + private volatile CountDownLatch exceptionLatch = new CountDownLatch(1); + + @Before + public void setup() { + this.handler = new AbstractReplyProducingMessageHandler() { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + final SettableListenableFuture future = new SettableListenableFuture(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + latch.await(10, TimeUnit.SECONDS); + switch (whichTest) { + case 0: + future.set("reply"); + break; + case 1: + future.setException(new RuntimeException("foo")); + break; + case 2: + future.setException(new MessagingException(requestMessage)); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + }); + return future; + } + + }; + this.handler.setAsyncReplySupported(true); + this.handler.setOutputChannel(this.output); + this.latch = new CountDownLatch(1); + Log logger = spy(TestUtils.getPropertyValue(this.handler, "logger", Log.class)); + new DirectFieldAccessor(this.handler).setPropertyValue("logger", logger); + doAnswer(new Answer() { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + failedCallbackMessage = (String) invocation.getArguments()[0]; + failedCallbackException = (Exception) invocation.getArguments()[1]; + exceptionLatch.countDown(); + return null; + } + + }).when(logger).error(anyString(), any(Throwable.class)); + } + + @Test + public void testGoodResult() { + this.whichTest = 0; + this.handler.handleMessage(new GenericMessage("foo")); + assertNull(this.output.receive(0)); + this.latch.countDown(); + Message received = this.output.receive(10000); + assertNotNull(received); + assertEquals("reply", received.getPayload()); + assertNull(this.failedCallbackException); + } + + @Test + public void testGoodResultWithReplyChannelHeader() { + this.whichTest = 0; + this.handler.setOutputChannel(null); + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("foo") + .setReplyChannel(replyChannel) + .build(); + this.handler.handleMessage(message); + assertNull(replyChannel.receive(0)); + this.latch.countDown(); + Message received = replyChannel.receive(10000); + assertNotNull(received); + assertEquals("reply", received.getPayload()); + assertNull(this.failedCallbackException); + } + + @Test + public void testGoodResultWithNoReplyChannelHeaderNoOutput() throws Exception { + this.whichTest = 0; + this.handler.setOutputChannel(null); + QueueChannel errorChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("foo").setErrorChannel(errorChannel).build(); + this.handler.handleMessage(message); + assertNull(this.output.receive(0)); + this.latch.countDown(); + Message errorMessage = errorChannel.receive(1000); + assertNotNull(errorMessage); + assertThat(errorMessage.getPayload(), instanceOf(DestinationResolutionException.class)); + assertEquals("no output-channel or replyChannel header available", + ((Throwable) errorMessage.getPayload()).getMessage()); + assertNull(((MessagingException) errorMessage.getPayload()).getFailedMessage()); + assertNotNull(this.failedCallbackException); + assertThat(this.failedCallbackException.getMessage(), containsString("or replyChannel header")); + } + + @Test + public void testRuntimeException() { + QueueChannel errorChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("foo") + .setErrorChannel(errorChannel) + .build(); + this.handler.handleMessage(message); + assertNull(this.output.receive(0)); + this.whichTest = 1; + this.latch.countDown(); + Message received = errorChannel.receive(10000); + assertNotNull(received); + assertThat(received.getPayload(), instanceOf(MessageHandlingException.class)); + assertEquals("foo", ((Throwable) received.getPayload()).getCause().getMessage()); + assertSame(message, ((MessagingException) received.getPayload()).getFailedMessage()); + assertNull(this.failedCallbackException); + } + + @Test + public void testMessagingException() { + QueueChannel errorChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("foo") + .setErrorChannel(errorChannel) + .build(); + this.handler.handleMessage(message); + assertNull(this.output.receive(0)); + this.whichTest = 2; + this.latch.countDown(); + Message received = errorChannel.receive(10000); + assertNotNull(received); + assertThat(received.getPayload(), instanceOf(MessagingException.class)); + assertSame(message, ((MessagingException) received.getPayload()).getFailedMessage()); + assertNull(this.failedCallbackException); + } + + @Test + public void testMessagingExceptionNoErrorChannel() throws Exception { + Message message = MessageBuilder.withPayload("foo") + .build(); + this.handler.handleMessage(message); + assertNull(this.output.receive(0)); + this.whichTest = 2; + this.latch.countDown(); + assertTrue(this.exceptionLatch.await(10, TimeUnit.SECONDS)); + assertNotNull(this.failedCallbackException); + assertThat(this.failedCallbackMessage, containsString("no 'errorChannel' header")); + } + + @Test + public void testGateway() throws Exception { + this.whichTest = 0; + GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean(Foo.class); + gpfb.setBeanFactory(mock(BeanFactory.class)); + DirectChannel input = new DirectChannel(); + gpfb.setDefaultRequestChannel(input); + gpfb.setDefaultReplyTimeout(10000L); + gpfb.afterPropertiesSet(); + Foo foo = (Foo) gpfb.getObject(); + this.handler.setOutputChannel(null); + EventDrivenConsumer consumer = new EventDrivenConsumer(input, this.handler); + consumer.afterPropertiesSet(); + consumer.start(); + this.latch.countDown(); + String result = foo.exchange("foo"); + assertEquals("reply", result); + } + + @Test + public void testGatewayWithException() throws Exception { + this.whichTest = 0; + GatewayProxyFactoryBean gpfb = new GatewayProxyFactoryBean(Foo.class); + gpfb.setBeanFactory(mock(BeanFactory.class)); + DirectChannel input = new DirectChannel(); + gpfb.setDefaultRequestChannel(input); + gpfb.setDefaultReplyTimeout(10000L); + gpfb.afterPropertiesSet(); + Foo foo = (Foo) gpfb.getObject(); + this.handler.setOutputChannel(null); + EventDrivenConsumer consumer = new EventDrivenConsumer(input, this.handler); + consumer.afterPropertiesSet(); + consumer.start(); + this.latch.countDown(); + try { + foo.exchange("foo"); + } + catch (MessagingException e) { + assertThat(e.getClass().getSimpleName(), equalTo("RuntimeException")); + assertThat(e.getMessage(), equalTo("foo")); + } + } + + private interface Foo { + + String exchange(String payload); + } + +}