INT-3254: AMQP inbound-gateway: add reply-timeout

JIRA: https://jira.springsource.org/browse/INT-3254

Attribute was missing from the namespace causing issues
with long-running consumers with async handoff.

Also fix typo in http, ampq and core schemas.

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -20,6 +20,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+import org.w3c.dom.Element;
+
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
@@ -28,13 +30,14 @@ import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
-import org.w3c.dom.Element;
 
 /**
  * Base class for inbound adapter parsers for the AMQP namespace.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
+ *
  * @since 2.1
  */
 abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefinitionParser {
@@ -100,9 +103,10 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
 			builder.addConstructorArgValue(listenerContainerBeanDef);
 		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
-		
+
 		IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
-		
+
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -24,18 +24,21 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.util.StringUtils;
 
 /**
  * Parser for the AMQP 'inbound-channel-adapter' element.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
+ *
  * @since 2.1
  */
 public class AmqpInboundChannelAdapterParser extends AbstractAmqpInboundAdapterParser {
 
 	AmqpInboundChannelAdapterParser() {
-		super("org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter");
+		super(AmqpInboundChannelAdapter.class.getName());
 	}
 
 	@Override

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -20,18 +20,20 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 
 /**
  * Parser for the AMQP 'inbound-gateway' element.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.1
  */
 public class AmqpInboundGatewayParser extends AbstractAmqpInboundAdapterParser {
 
 	AmqpInboundGatewayParser() {
-		super("org.springframework.integration.amqp.inbound.AmqpInboundGateway");
+		super(AmqpInboundGateway.class.getName());
 	}
 
 

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.0.xsd

@@ -261,6 +261,17 @@
 							]]></xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="reply-timeout" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+	Used to set the receiveTimeout on the underlying MessagingTemplate instance
+	(org.springframework.integration.core.MessagingTemplate) for receiving messages
+	from the reply channel. If not specified this property will default to "1000"
+	(1 second). Only applies if the container thread hands off to another thread
+	before the reply is sent.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml

@@ -11,7 +11,7 @@
 
 	<int:channel id="requests"/>
 
-	<si-amqp:inbound-gateway id="gateway" request-channel="requests" queue-names="test"
+	<si-amqp:inbound-gateway id="gateway" request-channel="requests" queue-names="test" reply-timeout="1234"
 			connection-factory="rabbitConnectionFactory" message-converter="testConverter"/>
 
 	<bean id="rabbitConnectionFactory" class="org.mockito.Mockito" factory-method="mock">

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -27,6 +27,7 @@ import org.junit.runner.RunWith;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessageProperties;
@@ -39,13 +40,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessagingException;
 import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.messaging.MessageHandler;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ReflectionUtils;
@@ -54,6 +55,8 @@ import org.springframework.util.ReflectionUtils;
  * @author Mark Fisher
  * @author Gunnar Hillert
  * @author Artem Bilan
+ * @author Gary Russell
+ *
  * @since 2.1
  */
 @ContextConfiguration
@@ -73,6 +76,8 @@ public class AmqpInboundGatewayParserTests {
 		assertSame(testConverter, templateConverter);
 		assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(gateway, "autoStartup"));
 		assertEquals(0, TestUtils.getPropertyValue(gateway, "phase"));
+		assertEquals(Long.valueOf(1234L), TestUtils.getPropertyValue(gateway, "replyTimeout", Long.class));
+		assertEquals(Long.valueOf(1234L), TestUtils.getPropertyValue(gateway, "messagingTemplate.receiveTimeout", Long.class));
 	}
 
 	@Test
@@ -87,6 +92,7 @@ public class AmqpInboundGatewayParserTests {
 	public void verifyUsageWithHeaderMapper() throws Exception{
 		DirectChannel requestChannel = context.getBean("requestChannel", DirectChannel.class);
 		requestChannel.subscribe(new MessageHandler() {
+			@Override
 			public void handleMessage(org.springframework.messaging.Message<?> siMessage)
 					throws MessagingException {
 				org.springframework.messaging.Message<?> replyMessage = MessageBuilder.fromMessage(siMessage).setHeader("bar", "bar").build();
@@ -103,15 +109,16 @@ public class AmqpInboundGatewayParserTests {
 		amqpTemplate = Mockito.spy(amqpTemplate);
 
 		Mockito.doAnswer(new Answer() {
-		      public Object answer(InvocationOnMock invocation) {
-		          Object[] args = invocation.getArguments();
-		          Message amqpReplyMessage = (Message) args[2];
-		          MessageProperties properties = amqpReplyMessage.getMessageProperties();
-		          assertEquals("bar", properties.getHeaders().get("bar"));
-		          return null;
-		      }})
-		 .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
-				 Mockito.any(Message.class), Mockito.any(CorrelationData.class));
+			@Override
+			public Object answer(InvocationOnMock invocation) {
+				Object[] args = invocation.getArguments();
+				Message amqpReplyMessage = (Message) args[2];
+				MessageProperties properties = amqpReplyMessage.getMessageProperties();
+				assertEquals("bar", properties.getHeaders().get("bar"));
+				return null;
+			}})
+				.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
+				Mockito.any(Message.class), Mockito.any(CorrelationData.class));
 		ReflectionUtils.setField(amqpTemplateField, gateway, amqpTemplate);
 
 		AbstractMessageListenerContainer mlc =

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

@@ -3876,7 +3876,7 @@
 						[OPTIONAL] Specifies the order in which this interceptor will be
 						added to the existing channel
 						interceptors (if any).
-						Negative value (e.g., -2) will signify BEFORE existing iinterceptors (if any). Positive
+						Negative value (e.g., -2) will signify BEFORE existing interceptors (if any). Positive
 						value (e.g., 2)
 						will signify AFTER existing interceptors (if any)
 					</xsd:documentation>

spring-integration-http/src/main/resources/org/springframework/integration/http/config/spring-integration-http-4.0.xsd

@@ -165,7 +165,7 @@
 					<xsd:attribute name="reply-timeout" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation><![CDATA[
-	Used to set the revceiveTimeout on the underlying MessagingTemplate instance
+	Used to set the receiveTimeout on the underlying MessagingTemplate instance
 	(org.springframework.integration.core.MessagingTemplate) for receiving messages
 	from the reply channel. If not specified this propery will default to "1000"
 	(1 second).