Merge pull request #196 from ghillert/INT-2245

Jdbc Outbound Gateways - Change "request-timeout" to "reply-timeout"

  For reference see: https://jira.springsource.org/browse/INT-2245

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParser.java

@@ -77,7 +77,8 @@ public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-messages-per-poll");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "keys-generated");
-
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
+		
 		String replyChannel = element.getAttribute("reply-channel");
 		if (StringUtils.hasText(replyChannel)) {
 			builder.addPropertyReference("outputChannel", replyChannel);

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParser.java

@@ -58,7 +58,8 @@ public class StoredProcOutboundGatewayParser extends AbstractConsumerEndpointPar
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, gatewayElement, "use-payload-as-parameter-source");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, gatewayElement, "sql-parameter-source-factory");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, gatewayElement, "skip-undeclared-results");
-
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, gatewayElement, "reply-timeout", "sendTimeout");
+		 
 		final ManagedList<BeanDefinition> procedureParameterList       = StoredProcParserUtils.getProcedureParameterBeanDefinitions(gatewayElement, parserContext);
 		final ManagedList<BeanDefinition> sqlParameterDefinitionList   = StoredProcParserUtils.getSqlParameterDefinitionBeanDefinitions(gatewayElement, parserContext);
 		final ManagedMap<String, BeanDefinition> returningResultsetMap = StoredProcParserUtils.getReturningResultsetBeanDefinitions(gatewayElement, parserContext);

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

@@ -418,7 +418,20 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="request-timeout" type="xsd:string" />
+					<xsd:attribute name="reply-timeout" type="xsd:string">
+					    <xsd:annotation>
+		                    <xsd:documentation><![CDATA[
+		                        Allows you to specify how long this gateway will wait for 
+		                        the reply message to be sent successfully before throwing 
+		                        an exception. Keep in mind that when sending to a 
+		                        DirectChannel, the invocation will occur in the sender's thread
+		                        so the failing of the send operation may be caused by other 
+		                        components further downstream. By default the Gateway will 
+		                        wait indefinitely. The value is specified in milliseconds.
+		                    ]]>
+		                    </xsd:documentation>
+		                </xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="keys-generated" type="xsd:boolean">
 						<xsd:annotation>
 							<xsd:appinfo>
@@ -901,7 +914,20 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="request-timeout" type="xsd:string" />
+			<xsd:attribute name="reply-timeout" type="xsd:string">
+			    <xsd:annotation>
+			        <xsd:documentation><![CDATA[
+			            Allows you to specify how long this gateway will wait for 
+			            the reply message to be sent successfully before throwing 
+			            an exception. Keep in mind that when sending to a 
+			            DirectChannel, the invocation will occur in the sender's thread
+			            so the failing of the send operation may be caused by other 
+			            components further downstream. By default the Gateway will 
+			            wait indefinitely. The value is specified in milliseconds.
+			        ]]>
+			        </xsd:documentation>
+			    </xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="order">
 				<xsd:annotation>
 					<xsd:documentation>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundAdapterWithPollerTest-context.xml

@@ -23,10 +23,12 @@
 		<int:poller fixed-rate="1000"/>
 	</int-jdbc:outbound-channel-adapter>
 
-	<jdbc:embedded-database type="H2" id="dataSource">
-		<jdbc:script location="org/springframework/integration/jdbc/config/outboundPollerSchema.sql" />
-	</jdbc:embedded-database>
-	
+	<jdbc:embedded-database type="H2" id="dataSource"/>
+		
+	<jdbc:initialize-database ignore-failures="DROPS">
+	    <jdbc:script location="org/springframework/integration/jdbc/config/outboundPollerSchema.sql" />
+	</jdbc:initialize-database>
+
 	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
 		<property name="dataSource" ref="dataSource"/>
 	</bean>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java

@@ -23,6 +23,7 @@ import javax.sql.DataSource;
 
 import org.junit.After;
 import org.junit.Test;
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -30,6 +31,8 @@ import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.jdbc.JdbcOutboundGateway;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
@@ -97,7 +100,7 @@ public class JdbcOutboundGatewayParserTests {
 		Map<String, ?> payload = (Map<String, ?>) reply.getPayload();
 		assertEquals(1, payload.get("updated"));
 	}
-
+	
 	@Test
 	public void testWithPoller() throws Exception{
 		ApplicationContext ac = new ClassPathXmlApplicationContext("JdbcOutboundGatewayWithPollerTest-context.xml", this.getClass());
@@ -115,6 +118,26 @@ public class JdbcOutboundGatewayParserTests {
 		Map<String, ?> payload = (Map<String, ?>) reply.getPayload();
 		assertEquals("bar", payload.get("name"));
 	}
+
+    @Test
+    public void testReplyTimeoutIsSet() throws Exception {
+        setUp("JdbcOutboundGatewayWithPollerTest-context.xml", getClass());
+
+        PollingConsumer outboundGateway = this.context.getBean("jdbcOutboundGateway", PollingConsumer.class);
+        
+        DirectFieldAccessor accessor = new DirectFieldAccessor(outboundGateway);
+        Object source = accessor.getPropertyValue("handler");
+        accessor = new DirectFieldAccessor(source);
+        source = accessor.getPropertyValue("messagingTemplate");
+        
+        MessagingTemplate messagingTemplate = (MessagingTemplate) source;
+
+        accessor = new DirectFieldAccessor(messagingTemplate);
+
+        Long  sendTimeout = (Long) accessor.getPropertyValue("sendTimeout");
+        assertEquals("Wrong sendTimeout", Long.valueOf(444L),  sendTimeout);
+        
+    }
 	
 	@After
 	public void tearDown() {

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPollerTest-context.xml

@@ -18,14 +18,17 @@
 		<int:queue />
 	</int:channel>
 
-	<int-jdbc:outbound-gateway query="select * from bazz where id=:headers[id]" update="insert into bazz (id, status, name) values (:headers[id], 0, :payload[foo])"
-		request-channel="target" reply-channel="output" data-source="dataSource" auto-startup="true">
+	<int-jdbc:outbound-gateway id="jdbcOutboundGateway" query="select * from bazz where id=:headers[id]" update="insert into bazz (id, status, name) values (:headers[id], 0, :payload[foo])"
+		request-channel="target" reply-channel="output" data-source="dataSource" auto-startup="true" reply-timeout="444">
 		<int:poller fixed-rate="1000"/>
 	</int-jdbc:outbound-gateway>
 
-	<jdbc:embedded-database type="H2" id="dataSource">
-		<jdbc:script location="org/springframework/integration/jdbc/config/outboundPollerSchema.sql" />
-	</jdbc:embedded-database>
+	<jdbc:embedded-database type="H2" id="dataSource"/>
+	
+	<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
+	    <jdbc:script location="org/springframework/integration/jdbc/config/outboundPollerSchema.sql" />
+	</jdbc:initialize-database>
+	
 	
 	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
 		<property name="dataSource" ref="dataSource"/>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParserTests.java

@@ -29,6 +29,7 @@ import org.junit.Test;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.jdbc.storedproc.PrimeMapper;
 import org.springframework.integration.jdbc.storedproc.ProcedureParameter;
@@ -61,6 +62,25 @@ public class StoredProcOutboundGatewayParserTests {
         assertEquals("Wrong stored procedure name", "GET_PRIME_NUMBERS",  storedProcedureName);
     }
 
+    @Test
+    public void testReplyTimeoutIsSet() throws Exception {
+        setUp("storedProcOutboundGatewayParserTest.xml", getClass());
+
+        DirectFieldAccessor accessor = new DirectFieldAccessor(this.outboundGateway);
+        Object source = accessor.getPropertyValue("handler");
+        accessor = new DirectFieldAccessor(source);
+        source = accessor.getPropertyValue("messagingTemplate");
+        
+        MessagingTemplate messagingTemplate = (MessagingTemplate) source;
+
+        accessor = new DirectFieldAccessor(messagingTemplate);
+
+        Long  sendTimeout = (Long) accessor.getPropertyValue("sendTimeout");
+        assertEquals("Wrong sendTimeout", Long.valueOf(555L),  sendTimeout);
+        
+    }
+
+    
     @Test
     public void testSkipUndeclaredResultsAttributeSet() throws Exception {
         setUp("storedProcOutboundGatewayParserTest.xml", getClass());

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundPollerSchema.sql

@@ -1,2 +1,5 @@
+drop table bazz;
+drop table foow;
+
 create table bazz(id varchar(100),status int,name varchar(20));
 create table foow(id varchar(100),status int,name varchar(20));

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/storedProcOutboundGatewayParserTest.xml

@@ -22,7 +22,7 @@
                                            skip-undeclared-results="false"
                                            order="2"
                                            reply-channel="replyChannel"
-                                           request-timeout="555"
+                                           reply-timeout="555"
                                            return-value-required="false">
 
         <int-jdbc:sql-parameter-definition name="username" direction="IN"    type="VARCHAR"/>