INT-1233: add sql-query-parameter-source support for JDBC inbound adapter

spring-integration-ip/.classpath

@@ -5,6 +5,11 @@
 	<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources"/>
 	<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
 	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
-	<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
+	<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="org.eclipse.jst.component.nondependency" value=""/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jst.j2ee.internal.module.container"/>
 	<classpathentry kind="output" path="target/classes"/>
 </classpath>

spring-integration-ip/.project

@@ -5,6 +5,11 @@
 	<projects>
 	</projects>
 	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.wst.common.project.facet.core.builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
 		<buildCommand>
 			<name>org.eclipse.jdt.core.javabuilder</name>
 			<arguments>
@@ -20,10 +25,18 @@
 			<arguments>
 			</arguments>
 		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.wst.validation.validationbuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
 	</buildSpec>
 	<natures>
+		<nature>org.eclipse.jem.workbench.JavaEMFNature</nature>
+		<nature>org.eclipse.wst.common.modulecore.ModuleCoreNature</nature>
 		<nature>org.springframework.ide.eclipse.core.springnature</nature>
 		<nature>org.maven.ide.eclipse.maven2Nature</nature>
 		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.wst.common.project.facet.core.nature</nature>
 	</natures>
 </projectDescription>

spring-integration-ip/.settings/org.eclipse.jdt.core.prefs

@@ -1,8 +1,9 @@
-#Wed May 26 00:58:46 CEST 2010
+#Sat Jul 10 12:27:30 BST 2010
 eclipse.preferences.version=1
 org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5
 org.eclipse.jdt.core.compiler.compliance=1.5
 org.eclipse.jdt.core.compiler.problem.annotationSuperInterface=warning
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
 org.eclipse.jdt.core.compiler.problem.autoboxing=ignore
 org.eclipse.jdt.core.compiler.problem.comparingIdentical=warning
 org.eclipse.jdt.core.compiler.problem.deadCode=warning
@@ -11,6 +12,7 @@ org.eclipse.jdt.core.compiler.problem.deprecationInDeprecatedCode=disabled
 org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=disabled
 org.eclipse.jdt.core.compiler.problem.discouragedReference=warning
 org.eclipse.jdt.core.compiler.problem.emptyStatement=ignore
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
 org.eclipse.jdt.core.compiler.problem.fallthroughCase=ignore
 org.eclipse.jdt.core.compiler.problem.fatalOptionalError=enabled
 org.eclipse.jdt.core.compiler.problem.fieldHiding=ignore

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java

@@ -94,6 +94,16 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
 	public void setSqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) {
 		this.sqlParameterSourceFactory = sqlParameterSourceFactory;
 	}
+	
+	/**
+	 * A source of parameters for the select query used for polling.
+	 * 
+	 * @param sqlQueryParameterSource the sql query parameter source to set
+	 */
+	public void setSqlQueryParameterSource(
+			SqlParameterSource sqlQueryParameterSource) {
+		this.sqlQueryParameterSource = sqlQueryParameterSource;
+	}
 
 	/**
 	 * Executes the query. If a query result set contains one or more rows, the Message

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

@@ -69,6 +69,7 @@ public class JdbcPollingChannelAdapterParser extends AbstractPollingInboundChann
 		builder.addConstructorArgValue(query);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-parameter-source-factory");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-query-parameter-source");
 		if (update!=null) {
 			builder.addPropertyValue("updateSql", update);
 		}

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

@@ -216,6 +216,21 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="sql-query-parameter-source" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+									Reference to a SqlParameterSource for the SELECT query used for polling. If
+									that query has placeholders (e.g. "SELECT * from FOO where KEY=:key") they
+									will be bound from this source by name.
+								</xsd:documentation>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.jdbc.core.namedparam.SqlParameterSource" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java

@@ -6,6 +6,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Types;
 import java.util.List;
 import java.util.Map;
 
@@ -14,6 +15,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.springframework.integration.core.Message;
 import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
 import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
@@ -28,7 +30,6 @@ public class JdbcPollingChannelAdapterIntegrationTests {
 
 	private SimpleJdbcTemplate jdbcTemplate;
 
-	
 	@Before
 	public void setUp() {
 		EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();
@@ -55,13 +56,49 @@ public class JdbcPollingChannelAdapterIntegrationTests {
 		assertTrue("Wrong payload type", payload instanceof List<?>);
 		List<?> rows = (List<?>) payload;
 		assertEquals("Wrong number of elements", 1, rows.size());
-		assertTrue("Returned row not a map", rows.get(0) instanceof Map<?,?>);
+		assertTrue("Returned row not a map", rows.get(0) instanceof Map<?, ?>);
 		Map<?, ?> row = (Map<?, ?>) rows.get(0);
 		assertEquals("Wrong id", 1, row.get("id"));
 		assertEquals("Wrong status", 2, row.get("status"));
 
 	}
-	
+
+	@Test
+	public void testParameterizedPollForListOfMapsNoUpdate() {
+		JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
+				this.embeddedDatabase,
+				"select * from item where status=:status");
+		adapter.setSqlQueryParameterSource(new SqlParameterSource() {
+
+			public boolean hasValue(String name) {
+				return "status".equals(name);
+			}
+
+			public Object getValue(String name) throws IllegalArgumentException {
+				return 2;
+			}
+
+			public String getTypeName(String name) {
+				return null;
+			}
+
+			public int getSqlType(String name) {
+				return Types.INTEGER;
+			}
+		});
+		this.jdbcTemplate.update("insert into item values(1,2)");
+		Message<Object> message = adapter.receive();
+		Object payload = message.getPayload();
+		assertTrue("Wrong payload type", payload instanceof List<?>);
+		List<?> rows = (List<?>) payload;
+		assertEquals("Wrong number of elements", 1, rows.size());
+		assertTrue("Returned row not a map", rows.get(0) instanceof Map<?, ?>);
+		Map<?, ?> row = (Map<?, ?>) rows.get(0);
+		assertEquals("Wrong id", 1, row.get("id"));
+		assertEquals("Wrong status", 2, row.get("status"));
+
+	}
+
 	@Test
 	public void testSimplePollForListWithRowMapperNoUpdate() {
 		JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
@@ -146,14 +183,13 @@ public class JdbcPollingChannelAdapterIntegrationTests {
 				countOfStatusTen);
 
 	}
-	
+
 	@Test
 	public void testEmptyPoll() {
 		JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
 				this.embeddedDatabase, "select * from item");
 		Message<Object> message = adapter.receive();
 		assertNull("Message received when no rows in table", message);
-		
 
 	}
 

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

@@ -17,6 +17,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.MessageChannelTemplate;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.core.Message;
+import org.springframework.jdbc.core.namedparam.AbstractSqlParameterSource;
 import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -71,7 +72,7 @@ public class JdbcPollingChannelAdapterParserTests {
 	}
 	
 	@Test
-	public void testParameterSourceInboundChannelAdapter(){
+	public void testParameterSourceFactoryInboundChannelAdapter(){
 		setUp("pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml", getClass());
 		this.jdbcTemplate.update("insert into item values(1,'',2)");
 		Message<?> message = channelTemplate.receive();
@@ -81,6 +82,14 @@ public class JdbcPollingChannelAdapterParserTests {
 		assertEquals("bar", list.get(0).get("NAME"));
 	}
 	
+	@Test
+	public void testParameterSourceInboundChannelAdapter(){
+		setUp("pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml", getClass());
+		this.jdbcTemplate.update("insert into item values(1,'',2)");
+		Message<?> message = channelTemplate.receive();
+		assertNotNull(message);
+	}
+	
 	@After
 	public void tearDown(){
 		if(appCtx != null){
@@ -104,5 +113,18 @@ public class JdbcPollingChannelAdapterParserTests {
 	protected void setupJdbcTemplate(){
 		this.jdbcTemplate = new SimpleJdbcTemplate(this.appCtx.getBean("dataSource",DataSource.class));
 	}
+	
+	public static class TestSqlParameterSource extends AbstractSqlParameterSource {
+
+		public Object getValue(String paramName)
+				throws IllegalArgumentException {
+			return 2;
+		}
+
+		public boolean hasValue(String paramName) {
+			return true;
+		}
+		
+	}
 
 }

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

@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration/jdbc"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    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/jdbc
+			http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd">
+	
+	<inbound-channel-adapter query="select * from item where status=:status" channel="target" 
+			data-source="dataSource" sql-query-parameter-source="parameterSource"/>
+		
+	<beans:import  resource="jdbcInboundChannelAdapterCommonConfig.xml" />
+	
+	<beans:bean id="parameterSource" class="org.springframework.integration.jdbc.config.JdbcPollingChannelAdapterParserTests$TestSqlParameterSource"/>
+
+</beans:beans>