diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index 69fe031a52..3816b2f7c8 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -26,7 +26,6 @@ import org.springframework.integration.handler.ReplyRequiredException; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.concurrent.ListenableFutureCallback; @@ -94,7 +93,7 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { } } logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend); - sendErrorMessage(exceptionToLogAndSend, this.requestMessage.getHeaders().getErrorChannel()); + sendErrorMessage(this.requestMessage, exceptionToLogAndSend); } } @@ -127,30 +126,7 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { } } else { - sendErrorMessage(exceptionToSend, this.requestMessage.getHeaders().getErrorChannel()); - } - } - - private void sendErrorMessage(Throwable ex, Object errorChannel) { - Throwable result = ex; - if (!(ex instanceof MessagingException)) { - result = new MessageHandlingException(this.requestMessage, ex); - } - if (errorChannel == null) { - logger.error("Async exception received and no 'errorChannel' header exists; cannot route " - + "exception to caller", result); - } - else { - try { - sendOutput(new ErrorMessage(result), errorChannel, true); - } - catch (Exception e) { - Exception exceptionToLog = e; - if (!(e instanceof MessagingException)) { - exceptionToLog = new MessageHandlingException(this.requestMessage, e); - } - logger.error("Failed to send async reply", exceptionToLog); - } + sendErrorMessage(this.requestMessage, exceptionToSend); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java index 8868eda7c5..6f461869d9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java @@ -46,6 +46,8 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; /** + * Factory bean to create and configure a {@link MessageHandler}. + * * @author Dave Syer * @author Oleg Zhurakousky * @author Gary Russell @@ -82,6 +84,8 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean channelResolver; + private Boolean asyncReplySupported; + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; @@ -97,14 +101,27 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean channelResolver) { this.channelResolver = channelResolver; } + /** + * Set the handler's output channel. + * @param outputChannel the output channel to set. + */ public void setOutputChannel(MessageChannel outputChannel) { this.outputChannel = outputChannel; } + /** + * Set the order in which the handler will be subscribed to its channel + * (when subscribable). + * @param order the order to set. + */ public void setOrder(Integer order) { this.order = order; } @@ -118,10 +135,28 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean adviceChain) { this.adviceChain = adviceChain; } + /** + * Currently only exposed on the service activator + * namespace. It's not clear that other endpoints would benefit from async support, + * but any subclass of {@link AbstractReplyProducingMessageHandler} can potentially + * return a {@code ListenableFuture}. + * @param asyncReplySupported the asyncReplySupported to set. + * @since 4.3 + */ + public void setAsyncReplySupported(Boolean asyncReplySupported) { + this.asyncReplySupported = asyncReplySupported; + } + /** * Sets the name of the handler component. * @@ -188,6 +223,12 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean requestMessage, Throwable ex) { + Object errorChannel = resolveErrorChannel(requestMessage.getHeaders()); + 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 and no default " + + "'errorChannel' found", result); + } + else { + try { + sendOutput(new ErrorMessage(result), 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); + } + } + } + + protected Object resolveErrorChannel(final MessageHeaders requestHeaders) { + Object errorChannel = requestHeaders.getErrorChannel(); + if (errorChannel == null) { + try { + errorChannel = getChannelResolver().resolveDestination("errorChannel"); + } + catch (DestinationResolutionException e) { + // ignore + } + } + return errorChannel; + } + } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd index 8aecead2e1..3ac59ea820 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd @@ -1227,6 +1227,16 @@ + + + and this flag is 'true', the calling + thread is released immediately. The remaining flow will run on the thread that completes + the future. If 'false' (default), the future will be sent as the payload of the result + message. This has no effect with any other type of return. + ]]> + + 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 index 2bfd3f6d42..7a0162e2a2 100644 --- 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 @@ -113,6 +113,7 @@ public class AsyncHandlerTests { }; this.handler.setAsyncReplySupported(true); this.handler.setOutputChannel(this.output); + this.handler.setBeanFactory(mock(BeanFactory.class)); this.latch = new CountDownLatch(1); Log logger = spy(TestUtils.getPropertyValue(this.handler, "logger", Log.class)); new DirectFieldAccessor(this.handler).setPropertyValue("logger", logger); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml index 508e3cc386..66bbfbafe2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml @@ -53,8 +53,24 @@ - + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index 92a610435a..ebd0f8ef38 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -16,13 +16,19 @@ package org.springframework.integration.handler; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +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.junit.Assert.fail; +import java.util.concurrent.atomic.AtomicReference; + import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; @@ -30,6 +36,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.support.MessageBuilder; @@ -38,8 +45,13 @@ import org.springframework.integration.util.StackTraceUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.SettableListenableFuture; /** * See INT-1688 for background. @@ -80,13 +92,23 @@ public class ServiceActivatorDefaultFrameworkMethodTests { @Autowired private TestMessageProcessor testMessageProcessor; + @Autowired + private MessageChannel asyncIn; + + @Autowired + private AsyncService asyncService; + + @Autowired + private PollableChannel errorChannel; + @Test public void testGateway() { QueueChannel replyChannel = new QueueChannel(); Message message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build(); this.gatewayTestInputChannel.send(message); Message reply = replyChannel.receive(0); - assertEquals("gatewayTestInputChannel,gatewayTestService,gateway,requestChannel,bridge,replyChannel", reply.getHeaders().get("history").toString()); + assertEquals("gatewayTestInputChannel,gatewayTestService,gateway,requestChannel,bridge,replyChannel", + reply.getHeaders().get("history").toString()); } @Test @@ -96,9 +118,11 @@ public class ServiceActivatorDefaultFrameworkMethodTests { this.replyingHandlerTestInputChannel.send(message); Message reply = replyChannel.receive(0); assertEquals("TEST", reply.getPayload()); - assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", reply.getHeaders().get("history").toString()); + assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", + reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", + "MethodInvokerHelper", st)); // close to the metal } @Test @@ -111,7 +135,8 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("optimizedRefReplyingHandlerTestInputChannel,optimizedRefReplyingHandlerTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", + "MethodInvokerHelper", st)); // close to the metal } @Test @@ -121,9 +146,11 @@ public class ServiceActivatorDefaultFrameworkMethodTests { this.replyingHandlerWithStandardMethodTestInputChannel.send(message); Message reply = replyChannel.receive(0); assertEquals("TEST", reply.getPayload()); - assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", reply.getHeaders().get("history").toString()); + assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", + reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", + "MethodInvokerHelper", st)); // close to the metal } @Test @@ -133,7 +160,8 @@ public class ServiceActivatorDefaultFrameworkMethodTests { this.replyingHandlerWithOtherMethodTestInputChannel.send(message); Message reply = replyChannel.receive(0); assertEquals("bar", reply.getPayload()); - assertEquals("replyingHandlerWithOtherMethodTestInputChannel,replyingHandlerWithOtherMethodTestService", reply.getHeaders().get("history").toString()); + assertEquals("replyingHandlerWithOtherMethodTestInputChannel,replyingHandlerWithOtherMethodTestService", + reply.getHeaders().get("history").toString()); } @Test @@ -161,7 +189,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { public void testFailOnDoubleReference() { try { new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "-fail-context.xml", - this.getClass()); + this.getClass()).close(); fail("Expected exception due to 2 endpoints referencing the same bean"); } catch (Exception e) { @@ -174,6 +202,68 @@ public class ServiceActivatorDefaultFrameworkMethodTests { } + @Test + public void testAsync() { + QueueChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build(); + this.asyncIn.send(message); + Message reply = replyChannel.receive(0); + assertNull(reply); + this.asyncService.future.set(this.asyncService.payload.toUpperCase()); + reply = replyChannel.receive(0); + assertNotNull(reply); + assertEquals("TEST", reply.getPayload()); + } + + @Test + public void testAsyncWithDirectReply() { + DirectChannel replyChannel = new DirectChannel(); + final AtomicReference> reply = new AtomicReference>(); + replyChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + reply.set(message); + } + }); + + Message message = MessageBuilder.withPayload("testing").setReplyChannel(replyChannel).build(); + this.asyncIn.send(message); + assertNull(reply.get()); + this.asyncService.future.set(this.asyncService.payload.toUpperCase()); + assertNotNull(reply.get()); + assertEquals("TESTING", reply.get().getPayload()); + } + + @Test + public void testAsyncError() { + QueueChannel errorChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("test").setErrorChannel(errorChannel).build(); + this.asyncIn.send(message); + this.asyncService.future.setException(new RuntimeException("intended")); + Message error = errorChannel.receive(0); + assertNotNull(error); + assertThat(error, instanceOf(ErrorMessage.class)); + assertThat(error.getPayload(), instanceOf(MessagingException.class)); + assertThat(((MessagingException) error.getPayload()).getCause(), instanceOf(RuntimeException.class)); + assertThat(((MessagingException) error.getPayload()).getCause().getMessage(), equalTo("intended")); + assertEquals("test", ((MessagingException) error.getPayload()).getFailedMessage().getPayload()); + } + + @Test + public void testAsyncErrorNoHeader() { + Message message = MessageBuilder.withPayload("test").build(); + this.asyncIn.send(message); + this.asyncService.future.setException(new RuntimeException("intended")); + Message error = this.errorChannel.receive(0); + assertNotNull(error); + assertThat(error, instanceOf(ErrorMessage.class)); + assertThat(error.getPayload(), instanceOf(MessagingException.class)); + assertThat(((MessagingException) error.getPayload()).getCause(), instanceOf(RuntimeException.class)); + assertThat(((MessagingException) error.getPayload()).getCause().getMessage(), equalTo("intended")); + assertEquals("test", ((MessagingException) error.getPayload()).getFailedMessage().getPayload()); + } + @SuppressWarnings("unused") private static class TestReplyingMessageHandler extends AbstractReplyProducingMessageHandler { @@ -189,7 +279,8 @@ public class ServiceActivatorDefaultFrameworkMethodTests { Exception e = new RuntimeException(); StackTraceElement[] st = e.getStackTrace(); // use this to test that StackTraceUtils works as expected and returns false - assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); + assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", + "MethodInvokerHelper", st)); return "bar"; } @@ -202,7 +293,8 @@ public class ServiceActivatorDefaultFrameworkMethodTests { public void handleMessage(Message requestMessage) { Exception e = new RuntimeException(); StackTraceElement[] st = e.getStackTrace(); - assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", + "MethodInvokerHelper", st)); // close to the metal } } @@ -221,4 +313,19 @@ public class ServiceActivatorDefaultFrameworkMethodTests { } } + private static class AsyncService { + + private volatile SettableListenableFuture future; + + private volatile String payload; + + @SuppressWarnings("unused") + public ListenableFuture process(String payload) { + this.future = new SettableListenableFuture(); + this.payload = payload; + return this.future; + } + + } + } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index bde5ca4160..e147ba06ec 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -977,7 +977,8 @@ _Optional (Defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE])_. This only applies if the `reply-channel` can block - such as a `QueueChannel` with a capacity limit that is currently full. Default: infinity. -<9> When `true`, the gateway will send an error message to the inbound message's `errorChannel` header if no reply +<9> When `true`, the gateway will send an error message to the inbound message's `errorChannel` header, +if present or otherwise to the default `errorChannel` (if available), when no reply message is received within the `AsyncRabbitTemplate`'s `receiveTimeout` property. Default: `true`. <10> The routing-key to use when sending Messages. @@ -986,8 +987,8 @@ Mutually exclusive with 'routing-key-expression'. _Optional_. -<11> A SpEL expression that is evaluated to determine the routing-key to use when sending Messages, with the message as the root object (e.g. -'payload.key'). +<11> A SpEL expression that is evaluated to determine the routing-key to use when sending Messages, +with the message as the root object (e.g. 'payload.key'). By default, this will be an empty String. Mutually exclusive with 'routing-key'. _Optional_. @@ -1036,6 +1037,7 @@ This allows "fail fast" detection of bad configuration, by logging an error mess When true (default), the connection is established (if it doesn't already exist because some other component established it) when the first message is sent. +Also see <> for more information. [IMPORTANT] .RabbitTemplate diff --git a/src/reference/asciidoc/service-activator.adoc b/src/reference/asciidoc/service-activator.adoc index 37ac9e9fb1..53d8e2d98e 100644 --- a/src/reference/asciidoc/service-activator.adoc +++ b/src/reference/asciidoc/service-activator.adoc @@ -106,3 +106,20 @@ For simple scenarios your _Service Activators_ do not even have to reference a b ---- In the above configuration our service logic is to simply multiply the payload value by 2, and SpEL lets us handle it relatively easy. + +[[async-service-activator]] +==== Asynchronous Service Activator + +The service activator is invoked by the calling thread; this would be some upstream thread if the input channel is a +`SubscribableChannel`, or a poller thread for a `PollableChannel`. +If the service returns a `ListenableFuture` the default action is to send that as the payload of the message sent +to the output (or reply) channel. +Starting with _version 4.3_, you can now set the `async` attribute to true (`setAsyncReplySupported(true)` when using +Java configuration). +If the service returns a `ListenableFuture` when this is true, the calling thread is released immediately, and the +reply message is sent on the thread (from within your service) that completes the future. +This is particularly advantageous for long-running services using a `PollableChannel` because the poller thread is +freed up to perform other services within the framework. + +If the service completes the future with an `Exception`, normal error processing will occur - an `ErrorMessage` is +sent to the `errorChannel` message header, if present or otherwise to the default `errorChannel` (if available). diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 88f4bca943..466197681e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -25,7 +25,9 @@ See <> for more information. [[x4.3-general]] === General Changes -==== Outbound Gateway within Chain +==== Core Changes + +===== Outbound Gateway within Chain Previously, it was possible to specify a `reply-channel` on an outbound gateway within a chain. It was completely ignored; the gateway's reply goes to the next chain element, or to the chain's output channel @@ -33,6 +35,11 @@ if the gateway is the last element. This condition is now detected and disallowed. If you have such configuration, simply remove the `reply-channel`. +===== Async Service Activator + +An option to make the Service Asynchronous has been added. +See <> for more information. + ==== Mail Changes The customizable `userFlag` added in 4.2.2 to provide customization of the flag used to denote that the mail has been