INT-3945: Async Service Activator

JIRA: https://jira.spring.io/browse/INT-3945

Polishing

Send errors to the default `errorChannel` (if available) and no `errorChannel`
header present.

Fix Test; Javadoc Polishing

Fix Test; Javadoc Polishing

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);
 			}
 		}
 

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<H extends MessageH
 
 	private DestinationResolver<MessageChannel> channelResolver;
 
+	private Boolean asyncReplySupported;
+
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
@@ -97,14 +101,27 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 		this.applicationEventPublisher = applicationEventPublisher;
 	}
 
+	/**
+	 * Set the handler's channel resolver.
+	 * @param channelResolver the channel resolver to set.
+	 */
 	public void setChannelResolver(DestinationResolver<MessageChannel> 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<H extends MessageH
 		return this.beanFactory;
 	}
 
+	/**
+	 * Set the advice chain to be configured within an
+	 * {@link AbstractReplyProducingMessageHandler} to advise just this local endpoint.
+	 * For other handlers, the advice chain is applied around the handler itself.
+	 * @param adviceChain the adviceChain to set.
+	 */
 	public void setAdviceChain(List<Advice> 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<H extends MessageH
 						+ (name == null ? "" : (", " + name)) + ".");
 				}
 			}
+			if (this.asyncReplySupported != null) {
+				if (actualHandler instanceof AbstractReplyProducingMessageHandler) {
+					((AbstractReplyProducingMessageHandler) actualHandler)
+							.setAsyncReplySupported(this.asyncReplySupported);
+				}
+			}
 			if (this.handler instanceof Orderable && this.order != null) {
 				((Orderable) this.handler).setOrder(this.order);
 			}

spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.java

@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
- * Base class for FactoryBeans that create MessageHandler instances.
+ * Base class for FactoryBeans that create standard MessageHandler instances.
  *
  * @author Mark Fisher
  * @author Alexander Peters
@@ -55,18 +55,34 @@ public abstract class AbstractStandardMessageHandlerFactoryBean
 
 	private volatile Expression expression;
 
+	/**
+	 * Set the target POJO for the message handler.
+	 * @param targetObject the target object.
+	 */
 	public void setTargetObject(Object targetObject) {
 		this.targetObject = targetObject;
 	}
 
+	/**
+	 * Set the method name for the message handler.
+	 * @param targetMethodName the target method name.
+	 */
 	public void setTargetMethodName(String targetMethodName) {
 		this.targetMethodName = targetMethodName;
 	}
 
+	/**
+	 * Set a SpEL expression to use.
+	 * @param expressionString the expression as a String.
+	 */
 	public void setExpressionString(String expressionString) {
 		this.expression = expressionParser.parseExpression(expressionString);
 	}
 
+	/**
+	 * Set a SpEL expression to use.
+	 * @param expression the expression.
+	 */
 	public void setExpression(Expression expression) {
 		this.expression = expression;
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-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.
@@ -16,6 +16,10 @@
 
 package org.springframework.integration.config.xml;
 
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ServiceActivatorFactoryBean;
 
 /**
@@ -37,4 +41,9 @@ public class ServiceActivatorParser extends AbstractDelegatingConsumerEndpointPa
 		return false;
 	}
 
+	@Override
+	void postProcess(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async", "asyncReplySupported");
+	}
+
 }

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

@@ -83,7 +83,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 	 *
 	 * @since 4.3
 	 */
-	protected final void setAsyncReplySupported(boolean asyncReplySupported) {
+	public final void setAsyncReplySupported(boolean asyncReplySupported) {
 		this.asyncReplySupported = asyncReplySupported;
 	}
 
@@ -204,27 +204,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 
 				@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(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);
-						}
-					}
+					sendErrorMessage(requestMessage, ex);
 				}
 			});
 		}
@@ -336,4 +316,41 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		return true;
 	}
 
+	protected void sendErrorMessage(final Message<?> 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;
+	}
+
 }

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd

@@ -1227,6 +1227,16 @@
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="async" type="xsd:string" use="optional">
+					<xsd:annotation>
+						<xsd:documentation><![CDATA[
+							If the service method returns a ListenableFuture<?> 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.
+						]]></xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

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);

spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml

@@ -53,8 +53,24 @@
 	<beans:bean id="testReplyingMessageHandler"
 		class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestReplyingMessageHandler"/>
 
-	<beans:bean id="testMessageProcessor" class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageProcessor">
+	<beans:bean id="testMessageProcessor"
+			class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$TestMessageProcessor">
 		<beans:property name="prefix" value="foo"/>
 	</beans:bean>
 
+	<channel id="asyncIn" />
+
+	<channel id="asyncOut">
+		<queue />
+	</channel>
+
+	<service-activator input-channel="asyncIn" ref="async" async="true" />
+
+	<beans:bean id="async"
+		class="org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests$AsyncService" />
+
+	<channel id="errorChannel">
+		<queue />
+	</channel>
+
 </beans:beans>

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<Message<?>> reply = new AtomicReference<Message<?>>();
+		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<String> future;
+
+		private volatile String payload;
+
+		@SuppressWarnings("unused")
+		public ListenableFuture<String> process(String payload) {
+			this.future = new SettableListenableFuture<String>();
+			this.payload = payload;
+			return this.future;
+		}
+
+	}
+
 }

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 <<async-service-activator>> for more information.
 
 [IMPORTANT]
 .RabbitTemplate

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).

src/reference/asciidoc/whats-new.adoc

@@ -25,7 +25,9 @@ See <<message-store>> 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 <<async-service-activator>> 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