diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParameterSourceFactory.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParameterSourceFactory.java index f055a04660..81f53c9912 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParameterSourceFactory.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParameterSourceFactory.java @@ -28,49 +28,81 @@ import org.springframework.jdbc.core.namedparam.SqlParameterSource; /** * A default implementation of {@link SqlParameterSourceFactory} which creates an {@link SqlParameterSource} according - * to the result of the data passed in. + * to the type of the data passed in. * + * * * @author Jonas Partner + * @author Dave Syer * @since 2.0 */ public class DefaultSqlParameterSourceFactory implements SqlParameterSourceFactory { private final Log logger = LogFactory.getLog(getClass()); - private final Map staticParameters; + private Map staticParameters; - private final String polledRowIdName = "id"; + private String rowIdName = "id"; - private final String updateIdsParamName = "idList"; + private String idsParamName = "idList"; public DefaultSqlParameterSourceFactory() { this.staticParameters = Collections.unmodifiableMap(new HashMap()); } - public DefaultSqlParameterSourceFactory(Map staticParameters) { - this.staticParameters = Collections.unmodifiableMap(staticParameters); + /** + * Name of the id property in the input elements when the input data is List. Defaults to "id". + * If the input is not a List then this value is ignored. + * @param rowIdName the name to set + */ + public void setRowIdName(String rowIdName) { + this.rowIdName = rowIdName; + } + + /** + * Name of the id list in the output parameters if the input is a List (default "idList"). If the input is not a + * List then this value is ignored. + * + * @param idsParamName the name to set + */ + public void setIdsParameterName(String idsParamName) { + this.idsParamName = idsParamName; + } + + /** + * If the input is a List or a Map, the output is a map parameter source, and in that case some static parameters + * can be added (default is empty). If the input is not a List or a Map then this value is ignored. + * + * @param staticParameters the static parameters to set + */ + public void setStaticParameters(Map staticParameters) { + this.staticParameters = staticParameters; } @SuppressWarnings("unchecked") - public SqlParameterSource createParameterSource(Object resultOfSelect) { + public SqlParameterSource createParameterSource(Object input) { SqlParameterSource toReturn; - if (resultOfSelect instanceof List) { + if (input instanceof List) { List ids = new ArrayList(); - for (Object rowObj : (List) resultOfSelect) { + for (Object rowObj : (List) input) { if (rowObj instanceof Map) { - ids.add(((Map) rowObj).get(this.polledRowIdName)); - } else { + ids.add(((Map) rowObj).get(this.rowIdName)); + } + else { DirectFieldAccessor accessor = new DirectFieldAccessor(rowObj); - if (accessor.isReadableProperty(this.polledRowIdName)) { - ids.add(accessor.getPropertyValue(this.polledRowIdName)); - } else { - logger.warn("No id field named '" + this.polledRowIdName + if (accessor.isReadableProperty(this.rowIdName)) { + ids.add(accessor.getPropertyValue(this.rowIdName)); + } + else { + logger.warn("No id field named '" + this.rowIdName + "' found for result of polled row. Update may not include all rows."); } } @@ -79,15 +111,17 @@ public class DefaultSqlParameterSourceFactory implements SqlParameterSourceFacto if (this.staticParameters != null) { thisParamSource.addValues(this.staticParameters); } - thisParamSource.addValue(this.updateIdsParamName, ids); + thisParamSource.addValue(this.idsParamName, ids); thisParamSource.getValue("idList"); toReturn = thisParamSource; - } else if (resultOfSelect instanceof Map) { - MapSqlParameterSource mapParameterSource = new MapSqlParameterSource((Map) resultOfSelect); + } + else if (input instanceof Map) { + MapSqlParameterSource mapParameterSource = new MapSqlParameterSource((Map) input); mapParameterSource.addValues(this.staticParameters); toReturn = mapParameterSource; - } else { - BeanPropertySqlParameterSource beanParameterSource = new BeanPropertySqlParameterSource(resultOfSelect); + } + else { + BeanPropertySqlParameterSource beanParameterSource = new BeanPropertySqlParameterSource(input); toReturn = beanParameterSource; } return toReturn; diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java index ea1b44f14e..e732e4bf50 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java @@ -51,7 +51,7 @@ public class JdbcPollingChannelAdapter implements MessageSource { private volatile String updateSql; - private volatile SqlParameterSourceFactory sqlParameterSourceFactoryForUpdate = new DefaultSqlParameterSourceFactory(); + private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new DefaultSqlParameterSourceFactory(); /** @@ -91,8 +91,8 @@ public class JdbcPollingChannelAdapter implements MessageSource { this.updatePerRow = updatePerRow; } - public void setSqlParameterSourceFactoryForUpdate(SqlParameterSourceFactory sqlParameterSourceFactoryForUpdate) { - this.sqlParameterSourceFactoryForUpdate = sqlParameterSourceFactoryForUpdate; + public void setSqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) { + this.sqlParameterSourceFactory = sqlParameterSourceFactory; } /** @@ -102,7 +102,7 @@ public class JdbcPollingChannelAdapter implements MessageSource { * this method will return null. */ public Message receive() { - Object payload = pollAndUpdate(); + Object payload = poll(); if (payload == null) { return null; } @@ -114,7 +114,7 @@ public class JdbcPollingChannelAdapter implements MessageSource { * Returns the rows returned by the select query. If a RowMapper * has been provided, the mapped results are returned. */ - private Object pollAndUpdate() { + private Object poll() { List payload; if (this.rowMapper != null) { payload = pollWithRowMapper(); @@ -140,8 +140,8 @@ public class JdbcPollingChannelAdapter implements MessageSource { private void executeUpdateQuery(Object obj) { SqlParameterSource updateParamaterSource = null; - if (this.sqlParameterSourceFactoryForUpdate != null) { - updateParamaterSource = this.sqlParameterSourceFactoryForUpdate.createParameterSource(obj); + if (this.sqlParameterSourceFactory != null) { + updateParamaterSource = this.sqlParameterSourceFactory.createParameterSource(obj); this.jdbcOperations.update(this.updateSql, updateParamaterSource); } else { diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/SqlParameterSourceFactory.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/SqlParameterSourceFactory.java index 85c89f49da..818976f240 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/SqlParameterSourceFactory.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/SqlParameterSourceFactory.java @@ -19,9 +19,8 @@ package org.springframework.integration.jdbc; import org.springframework.jdbc.core.namedparam.SqlParameterSource; /** - * Collaborator for {@link JdbcPollingChannelAdapter} which allows creation of - * instances of {@link SqlParameterSource} for use in updates to be created - * according to the result of the poll. + * Collaborator for JDBC adapters which allows creation of + * instances of {@link SqlParameterSource} for use in update operations. * * @author Jonas Partner * @since 2.0 @@ -30,8 +29,8 @@ public interface SqlParameterSourceFactory { /** * Return a new {@link SqlParameterSource}. - * @param resultOfSelect the result of the preceding poll operation + * @param input the raw message or query result to be transformed into a SqlParameterSource */ - public SqlParameterSource createParameterSource(Object resultOfSelect); + public SqlParameterSource createParameterSource(Object input); } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java index 047726a8ea..3de068f3b9 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java @@ -18,6 +18,7 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.util.StringUtils; import org.w3c.dom.Element; @@ -60,6 +61,7 @@ public class JdbcMessageHandlerParser extends AbstractOutboundChannelAdapterPars } else { builder.addConstructorArgReference(jdbcOperationsRef); } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-parameter-source-factory"); builder.addConstructorArgValue(query); return builder.getBeanDefinition(); } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java index b02bd94aa6..5d2d044c06 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java @@ -67,6 +67,7 @@ public class JdbcPollingChannelAdapterParser extends AbstractPollingInboundChann } builder.addConstructorArgValue(query); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-parameter-source-factory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "update", "updateSql"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "update-per-row"); return BeanDefinitionReaderUtils.registerWithGeneratedName( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd index 6e2a304fcd..80f1533637 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd @@ -1,15 +1,12 @@ - - + Reference to a data source to use to access - the database. Either this or the jdbc-operations must be + the database. Either this or the jdbc-operations + must be specified (but not both). @@ -51,8 +49,7 @@ specified (but not both). - + @@ -62,7 +59,8 @@ Unique string to use as a partition for the data in this store, so that - multiple instances can share the same + multiple instances can + share the same database tables. The default is "DEFAULT". @@ -72,7 +70,8 @@ Prefix for the table names in the database - (e.g. so that a schema can be specified, or to avoid a clash + (e.g. so that a schema can be specified, or to avoid + a clash with other tables). The default is "INT_". @@ -86,8 +85,7 @@ ]]> - + @@ -105,64 +103,74 @@ - - - - - - - Reference to a data source to use to access - the - database. Either this or the simple-jdbc-operations must be - specified (but not both). - - - - - - - - - - - - - Reference to a JdbcOperations. Either - this or - the data-source must be - specified (but not both). - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + A select query to execute when a message is polled. In general the query can return multiple + rows, because the result will be a List (of type determined by the row mapper). + + + + + + + + + Reference to a row mapper to use to convert JDBC result set rows to message payloads. + Optional + with default that maps + result set row to a map (column name to column value). Other simple + use cases can + be handled + with out-of-the box implementations from Spring JDBC. Others require a custom row + mapper. + + + + + + + + + + + + An update query to execute when a message is polled. If the poll is in a transaction then the + update will roll back if the transaction does. + + + + + + + + + Flag to indicate whether the update query should be executed per message, or per row (in the + case that a message contains multiple rows). + + + + + + + + + Channel to which polled messages will be sent. + + + + + + + + + @@ -174,59 +182,88 @@ - - - - Reference to a data source to use to access - the - database. Either this or the simple-jdbc-operations must be - specified (but not both). - - - - - - - - - - - - - Reference to a JdbcOperations. Either - this or - the data-source must be - specified (but not both). + + + + + + + An SQL update query to execute (INSERT, UPDATE + or DELETE). Bean properties of the outgoing + message can be + referenced in named parameters, e.g. "INSERT into FOOS (ID, NAME) values (:headers[business.key], + :payload)". More complex requirements can be implemented by - - - - - - - - - - - An SQL update query to execute (INSERT, UPDATE - or DELETE). Bean properties of the outgoing message can be - referenced in named parameters, e.g. "INSERT into FOOS (ID, NAME) values (:headers[business.key], :payload)" - - - - - - - - - - - - - + + + + + + + + Channel from which messages will be output. When a message is sent to this channel it will + cause the query to be executed. + + + + + + + + + + + + + + Reference to a data source to use to access + the + database. Either this or the + simple-jdbc-operations + must be + specified (but not both). + + + + + + + + + + + + + Reference to a JdbcOperations. Either + this or + the data-source must be + specified (but not both). + + + + + + + + + + + + Reference to a SqlParameterSourceFactory. For an inbound adapter the input is the result of the + query, and for an outbound adapter the input is the whole outgoing message. The default factory creates a bean + property parameter source for a generic input (like a Message), and treats a List in a special way: the List is + assumed to contain entities with a field called "id" and these are collected and copied to a field in the + parameter source called "idList". + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml similarity index 100% rename from spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml rename to spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java index c09a88c4e6..862e13bc92 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java @@ -25,7 +25,7 @@ public class JdbcMessageHandlerParserTests { private ConfigurableApplicationContext context; @Test - public void testSimpleInboundChannelAdapter(){ + public void testSimpleOutboundChannelAdapter(){ setUp("handlingWithJdbcOperationsJdbcOutboundChannelAdapterTest.xml", getClass()); Message message = MessageBuilder.withPayload("foo").setHeader("business.key", "FOO").build(); channel.send(message); @@ -35,7 +35,7 @@ public class JdbcMessageHandlerParserTests { } @Test - public void testDollarHeaderInboundChannelAdapter(){ + public void testDollarHeaderOutboundChannelAdapter(){ setUp("handlingDollarHeaderJdbcOutboundChannelAdapterTest.xml", getClass()); Message message = MessageBuilder.withPayload("foo").build(); channel.send(message); @@ -45,13 +45,23 @@ public class JdbcMessageHandlerParserTests { } @Test - public void testMapPayloadInboundChannelAdapter(){ + public void testMapPayloadOutboundChannelAdapter(){ setUp("handlingMapPayloadJdbcOutboundChannelAdapterTest.xml", getClass()); Message message = MessageBuilder.withPayload(Collections.singletonMap("foo", "bar")).build(); channel.send(message); Map map = this.jdbcTemplate.queryForMap("SELECT * from FOOS"); assertEquals("Wrong id", message.getHeaders().getId().toString(), map.get("ID")); - assertEquals("Wrong id", "bar", map.get("name")); + assertEquals("Wrong name", "bar", map.get("name")); + } + + @Test + public void testParameterSourceOutboundChannelAdapter(){ + setUp("handlingParameterSourceJdbcOutboundChannelAdapterTest.xml", getClass()); + Message message = MessageBuilder.withPayload("foo").build(); + channel.send(message); + Map map = this.jdbcTemplate.queryForMap("SELECT * from FOOS"); + assertEquals("Wrong id", message.getHeaders().getId().toString(), map.get("ID")); + assertEquals("Wrong name", "foo", map.get("name")); } @After diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java index e44d9866db..dc48c23dda 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java @@ -1,10 +1,12 @@ package org.springframework.integration.jdbc.config; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import java.util.List; +import java.util.Map; import javax.sql.DataSource; @@ -24,18 +26,16 @@ public class JdbcPollingChannelAdapterParserTests { final long receiveTimeout = 5000; - SimpleJdbcTemplate jdbcTemplate; - - MessageChannelTemplate channelTemplate; - - ConfigurableApplicationContext appCtx; - + private SimpleJdbcTemplate jdbcTemplate; + private MessageChannelTemplate channelTemplate; + private ConfigurableApplicationContext appCtx; + @Test public void testSimpleInboundChannelAdapter(){ setUp("pollingForMapJdbcInboundChannelAdapterTest.xml", getClass()); - this.jdbcTemplate.update("insert into item values(1,2)"); + this.jdbcTemplate.update("insert into item values(1,'',2)"); Message message = channelTemplate.receive(); assertNotNull("No message found ", message); assertTrue("Wrong payload type expected instance of List", message.getPayload() instanceof List); @@ -45,7 +45,7 @@ public class JdbcPollingChannelAdapterParserTests { @Test public void testSimpleInboundChannelAdapterWithUpdate(){ setUp("pollingForMapJdbcInboundChannelAdapterWithUpdateTest.xml", getClass()); - this.jdbcTemplate.update("insert into item values(1,2)"); + this.jdbcTemplate.update("insert into item values(1,'',2)"); Message message = channelTemplate.receive(); assertNotNull(message); message = channelTemplate.receive(); @@ -55,11 +55,22 @@ public class JdbcPollingChannelAdapterParserTests { @Test public void testExtendedInboundChannelAdapter(){ setUp("pollingWithJdbcOperationsJdbcInboundChannelAdapterTest.xml", getClass()); - this.jdbcTemplate.update("insert into item values(1,2)"); + this.jdbcTemplate.update("insert into item values(1,'',2)"); Message message = channelTemplate.receive(); assertNotNull(message); } + @Test + public void testParameterSourceInboundChannelAdapter(){ + setUp("pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml", getClass()); + this.jdbcTemplate.update("insert into item values(1,'',2)"); + Message message = channelTemplate.receive(); + assertNotNull(message); + List> list = jdbcTemplate.queryForList("SELECT * FROM item WHERE status=1"); + assertEquals(1, list.size()); + assertEquals("bar", list.get(0).get("NAME")); + } + @After public void tearDown(){ if(appCtx != null){ diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingParameterSourceJdbcOutboundChannelAdapterTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingParameterSourceJdbcOutboundChannelAdapterTest.xml new file mode 100644 index 0000000000..9273f9228d --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingParameterSourceJdbcOutboundChannelAdapterTest.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/inboundSchema.sql b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/inboundSchema.sql index 1f0246d3b8..b223e45b9d 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/inboundSchema.sql +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/inboundSchema.sql @@ -1 +1 @@ -create table item(id int,status int); \ No newline at end of file +create table item(id int, name varchar(20), status int); \ No newline at end of file diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml new file mode 100644 index 0000000000..fbe763a5af --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + +