Merge pull request #148 from markfisher/INT-2196

* INT-2196:
  polishing
  enable asyncExecutor to be configured on gateways

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

@@ -41,7 +41,7 @@ import org.springframework.util.xml.DomUtils;
 public class GatewayParser extends AbstractSimpleBeanDefinitionParser {
 
 	private static String[] referenceAttributes = new String[] {
-		"default-request-channel", "default-reply-channel", "error-channel", "message-mapper"
+		"default-request-channel", "default-reply-channel", "error-channel", "message-mapper", "async-executor"
 	};
 
 	private static String[] innerAttributes = new String[] {
@@ -81,6 +81,7 @@ public class GatewayParser extends AbstractSimpleBeanDefinitionParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "request-timeout", "defaultRequestTimeout");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "defaultReplyTimeout");		
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "async-executor");
 	}
 
 	private void postProcessGateway(BeanDefinitionBuilder builder, Element element) {

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

@@ -594,6 +594,18 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="async-executor" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+							<![CDATA[
+					Provide a reference to an implementation of java.util.concurrent.Executor
+					to use for any of the interface methods that have a Future return type.
+					This Executor will only be used for those async methods; the sync methods
+					will be invoked in the caller's thread.
+							]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 

spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 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.
@@ -18,18 +18,23 @@ package org.springframework.integration.config.xml;
 
 import static org.junit.Assert.assertEquals;
 
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
-
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.gateway.TestService;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.scheduling.annotation.AsyncResult;
 
 /**
  * @author Mark Fisher
@@ -67,6 +72,19 @@ public class GatewayParserTests {
 		assertEquals("foo", result);		
 	}
 
+	@Test
+	public void testAsyncGateway() throws Exception {
+		ApplicationContext context = new ClassPathXmlApplicationContext("gatewayParserTests.xml", this.getClass());
+		PollableChannel requestChannel = (PollableChannel) context.getBean("requestChannel");
+		MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel");
+		this.startResponder(requestChannel, replyChannel);
+		TestService service = context.getBean("async", TestService.class);
+		Future<Message<?>> result = service.async("foo");
+		Message<?> reply = result.get(1, TimeUnit.SECONDS);
+		assertEquals("foo", reply.getPayload());
+		assertEquals("testExecutor", reply.getHeaders().get("executor"));
+	}
+
 
 	private void startResponder(final PollableChannel requestChannel, final MessageChannel replyChannel) {
 		Executors.newSingleThreadExecutor().execute(new Runnable() {
@@ -79,4 +97,32 @@ public class GatewayParserTests {
 		});
 	}
 
+
+	@SuppressWarnings("unused")
+	private static class TestExecutor extends SimpleAsyncTaskExecutor implements BeanNameAware {
+
+		private static final long serialVersionUID = 1L;
+
+		private volatile String beanName;
+
+		public void setBeanName(String beanName) {
+			this.beanName = beanName;
+		}
+
+		@Override
+		@SuppressWarnings({"rawtypes", "unchecked"})
+		public <T> Future<T> submit(Callable<T> task) {
+			try {
+				Future<?> result = super.submit(task);
+				Message<?> message = (Message<?>) result.get(1, TimeUnit.SECONDS);
+				Message<?> modifiedMessage = MessageBuilder.fromMessage(message)
+						.setHeader("executor", this.beanName).build();
+				return new AsyncResult(modifiedMessage);
+			}
+			catch (Exception e) {
+				throw new IllegalStateException("unexpected exception in testExecutor", e);
+			}
+		}
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/config/xml/gatewayParserTests.xml

@@ -1,11 +1,11 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans:beans xmlns="http://www.springframework.org/schema/integration"
 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:beans="http://www.springframework.org/schema/beans"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd
 			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd">
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
 
 	<channel id="requestChannel">
 		<queue capacity="100"/>
@@ -29,8 +29,16 @@
 	         default-request-channel="requestChannel"
 	         default-reply-channel="replyChannel"
 	         default-reply-timeout="5000"/>
-	         
+
+	<gateway id="async"
+			 service-interface="org.springframework.integration.gateway.TestService"
+			 default-request-channel="requestChannel"
+			 default-reply-channel="replyChannel"
+			 async-executor="testExecutor"/>
+	
 	<!-- no assertions for this. The fact that this config does not result in error is sufficient -->
 	<gateway default-request-channel="nullChannel"/>
 
+	<beans:bean id="testExecutor" class="org.springframework.integration.config.xml.GatewayParserTests$TestExecutor"/>
+
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.gateway;
 
+import java.util.concurrent.Future;
+
 import org.springframework.integration.Message;
 import org.springframework.integration.annotation.Payload;
 
@@ -42,4 +44,6 @@ public interface TestService {
 	@Payload("#method + #args.length")
 	String requestReplyWithPayloadAnnotation();
 
+	Future<Message<?>> async(String s);
+
 }