diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParamterSourceFactory.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParamterSourceFactory.java new file mode 100644 index 0000000000..1604b561c5 --- /dev/null +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParamterSourceFactory.java @@ -0,0 +1,109 @@ +/* + * Copyright 2002-2008 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.jdbc; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.jdbc.core.namedparam.BeanPropertySqlParameterSource; +import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; +import org.springframework.jdbc.core.namedparam.SqlParameterSource; + +/** + * A default implementation of {@link SqlParamterSourceFactory} which creates an + * {@link SqlParameterSource} according to the result of the polled data passed + * in + * + * Where the result of the poll is a List a list of ids generated by looking for + * a map entry or bean property named by default 'id'. The resulting {@link SqlParameterSource} + * contains this list under a default key of 'idList'. + * + * Where the result of the poll is a {@link Map} this is wrapped in an instance of {@link MapSqlParameterSource} + * Otherwise the result is wrapped in an instance of {@link BeanPropertySqlParameterSource} + * + * @author Jonas Partner + * + */ +public class DefaultSqlParamterSourceFactory implements + SqlParamterSourceFactory { + + private final Log logger = LogFactory.getLog(getClass()); + + private final Map staticParameters; + + private volatile String polledRowIdName = "id"; + + private volatile String updateIdsParamName = "idList"; + + public DefaultSqlParamterSourceFactory() { + this.staticParameters = Collections + .unmodifiableMap(new HashMap()); + } + + public DefaultSqlParamterSourceFactory(Map staticParameters) { + this.staticParameters = Collections.unmodifiableMap(staticParameters); + } + + public SqlParameterSource createParamterSource(Object resultOfSelect) { + SqlParameterSource toReturn; + if (resultOfSelect instanceof List) { + List ids = new ArrayList(); + for (Object rowObj : (List) resultOfSelect) { + if (rowObj instanceof Map) { + ids.add(((Map) rowObj).get(this.polledRowIdName)); + } 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 + + " found for result of polled row. Update may not include all rows"); + } + } + } + MapSqlParameterSource thisParamSource = new MapSqlParameterSource(); + if (this.staticParameters != null) { + thisParamSource.addValues(this.staticParameters); + } + thisParamSource.addValue(this.updateIdsParamName, ids); + thisParamSource.getValue("idList"); + toReturn = thisParamSource; + } else if (resultOfSelect instanceof Map) { + MapSqlParameterSource mapParameterSource = new MapSqlParameterSource( + (Map) resultOfSelect); + mapParameterSource.addValues(this.staticParameters); + toReturn = mapParameterSource; + } else { + BeanPropertySqlParameterSource beanParameterSource = new BeanPropertySqlParameterSource( + resultOfSelect); + toReturn = beanParameterSource; + } + return toReturn; + } + +} diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java index 55a653c4ba..cef13541fd 100644 --- a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java @@ -16,6 +16,11 @@ package org.springframework.integration.jdbc; +import java.util.List; +import java.util.Map; + +import javax.sql.DataSource; + import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; @@ -31,115 +36,154 @@ import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; -import javax.sql.DataSource; -import java.util.List; - /** - * A polling channel adapter that creates messages from the payload returned by executing a select query - * Optionally an update can be executed after the select in order to update processed rows - * + * A polling channel adapter that creates messages from the payload returned by + * executing a select query Optionally an update can be executed after the + * select in order to update processed rows + * * @author Jonas Partner */ -public class JdbcPollingChannelAdapter implements MessageSource, InitializingBean { +public class JdbcPollingChannelAdapter implements MessageSource, + InitializingBean { - private final SimpleJdbcOperations jdbcOperations; + private final SimpleJdbcOperations jdbcOperations; - private final String selectQuery; + private final String selectQuery; - private volatile RowMapper rowMapper; + private volatile RowMapper rowMapper; - private volatile String updateQuery; + private volatile SqlParameterSource sqlQueryParameterSource; - private volatile SqlParameterSource sqlQueryParameterSource; + private volatile TransactionDefinition transactionDefinition; - private volatile TransactionDefinition transactionDefinition; + private volatile TransactionTemplate transactionTemplate; - private volatile TransactionTemplate transactionTemplate; + private volatile PlatformTransactionManager platformTransactionManager; - private volatile PlatformTransactionManager platformTransactionManager; + private volatile boolean updatePerRow = false; - public JdbcPollingChannelAdapter(DataSource dataSource, String selectQuery) { - this.jdbcOperations = new SimpleJdbcTemplate(dataSource); - this.selectQuery = selectQuery; - } + private volatile String updateQuery; - public JdbcPollingChannelAdapter(SimpleJdbcOperations jdbcOperations, String selectQuery) { - this.jdbcOperations = jdbcOperations; - this.selectQuery = selectQuery; - } + private volatile SqlParamterSourceFactory sqlParameterSourceFactoryForUpdate = new DefaultSqlParamterSourceFactory(); - public void setTransactionDefinition(TransactionDefinition transactionDefinition) { - this.transactionDefinition = transactionDefinition; - } + public JdbcPollingChannelAdapter(DataSource dataSource, String selectQuery) { + this.jdbcOperations = new SimpleJdbcTemplate(dataSource); + this.selectQuery = selectQuery; + } - public void setTransactionManager(PlatformTransactionManager transactionManager) { - this.platformTransactionManager = platformTransactionManager; - } + public JdbcPollingChannelAdapter(SimpleJdbcOperations jdbcOperations, + String selectQuery) { + this.jdbcOperations = jdbcOperations; + this.selectQuery = selectQuery; + } - public void setRowMapper(RowMapper rowMapper) { - this.rowMapper = rowMapper; - } + public void setTransactionDefinition( + TransactionDefinition transactionDefinition) { + this.transactionDefinition = transactionDefinition; + } - public void setUpdateQuery(String updateQuery) { - this.updateQuery = updateQuery; - } + public void setTransactionManager( + PlatformTransactionManager platformTransactionManager) { + this.platformTransactionManager = platformTransactionManager; + } + public void setRowMapper(RowMapper rowMapper) { + this.rowMapper = rowMapper; + } - public void afterPropertiesSet() throws Exception { - if (this.transactionDefinition == null) { - this.transactionDefinition = new DefaultTransactionDefinition(); - } + public void setUpdateQuery(String updateQuery) { + this.updateQuery = updateQuery; + } + + public void setUpdatePerRow(boolean updatePerRow){ + this.updatePerRow = updatePerRow; + } - if (this.platformTransactionManager != null) { - this.transactionTemplate = new TransactionTemplate(this.platformTransactionManager, this.transactionDefinition); - } - } + public void setSqlParameterSourceFactoryForUpdate( + SqlParamterSourceFactory sqlParameterSourceFactoryForUpdate) { + this.sqlParameterSourceFactoryForUpdate = sqlParameterSourceFactoryForUpdate; + } - public Message receive() { - Object payload = null; - if (this.transactionTemplate != null){ - payload = this.transactionTemplate.execute(new TransactionCallback(){ - public Object doInTransaction(TransactionStatus status) { - return pollAndUpdate(); - } - }); - } else { - payload = pollAndUpdate(); - } - return MessageBuilder.withPayload(payload).build(); - } + public void afterPropertiesSet() throws Exception { + if (this.transactionDefinition == null) { + this.transactionDefinition = new DefaultTransactionDefinition(); + } - protected Object pollAndUpdate(){ - List payload; - if (this.rowMapper != null) { - payload = pollWithRowMapper(); - } else { - payload = this.jdbcOperations.queryForList(this.selectQuery, this.sqlQueryParameterSource); - } + if (this.platformTransactionManager != null) { + this.transactionTemplate = new TransactionTemplate( + this.platformTransactionManager, this.transactionDefinition); + } + } - return payload; + public Message receive() { + Object payload = null; + if (this.transactionTemplate != null) { + payload = this.transactionTemplate + .execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return pollAndUpdate(); + } + }); + } else { + payload = pollAndUpdate(); + } + return MessageBuilder.withPayload(payload).build(); + } - } + protected Object pollAndUpdate() { + List payload; + if (this.rowMapper != null) { + payload = pollWithRowMapper(); + } else { + payload = this.jdbcOperations.queryForList(this.selectQuery, + this.sqlQueryParameterSource); + } + if (updateQuery != null) { + if (this.updatePerRow) { + for (Object row : payload) { + executeUpdateQuery(row); + } + } else { + executeUpdateQuery(payload); + } + } + return payload; - protected List pollWithRowMapper(){ - List payload = null; - if(this.sqlQueryParameterSource != null){ - payload = this.jdbcOperations.query(this.selectQuery, this.rowMapper, this.sqlQueryParameterSource); - } else { - payload = this.jdbcOperations.query(this.selectQuery, this.rowMapper); - } - return payload; - } + } - protected Object pollForListOfMap(){ - List payload = null; - if(this.sqlQueryParameterSource != null){ - payload = this.jdbcOperations.queryForList(this.selectQuery, this.sqlQueryParameterSource); - } else { - payload = this.jdbcOperations.queryForList(this.selectQuery); - } - return payload; - } + protected void executeUpdateQuery(Object obj) { + SqlParameterSource updateParamaterSource = null; + if (this.sqlParameterSourceFactoryForUpdate != null) { + + updateParamaterSource = this.sqlParameterSourceFactoryForUpdate + .createParamterSource(obj); + this.jdbcOperations.update(this.updateQuery, updateParamaterSource); + } else { + this.jdbcOperations.update(this.updateQuery); + } + } + protected List pollWithRowMapper() { + List payload = null; + if (this.sqlQueryParameterSource != null) { + payload = this.jdbcOperations.query(this.selectQuery, + this.rowMapper, this.sqlQueryParameterSource); + } else { + payload = this.jdbcOperations.query(this.selectQuery, + this.rowMapper); + } + return payload; + } -} + protected List> pollForListOfMap() { + List> payload = null; + if (this.sqlQueryParameterSource != null) { + payload = this.jdbcOperations.queryForList(this.selectQuery, + this.sqlQueryParameterSource); + } else { + payload = this.jdbcOperations.queryForList(this.selectQuery); + } + return payload; + } + +} \ No newline at end of file diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/SqlParamterSourceFactory.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/SqlParamterSourceFactory.java new file mode 100644 index 0000000000..09f5096bfd --- /dev/null +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/SqlParamterSourceFactory.java @@ -0,0 +1,38 @@ +/* + * Copyright 2002-2008 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +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 + * + * @author Jonas Partner + * + */ +public interface SqlParamterSourceFactory { + + /** + * Return a new {@link SqlParameterSource} + * @param obj the result of the preceeding poll operation + * @return + */ + public SqlParameterSource createParamterSource(Object obj); + +} diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java index 43970903ce..23a4c776e5 100644 --- a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java @@ -1,6 +1,7 @@ package org.springframework.integration.jdbc; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import java.sql.ResultSet; import java.sql.SQLException; @@ -9,7 +10,6 @@ import java.util.Map; import org.junit.After; import org.junit.Before; -import org.junit.BeforeClass; import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.jdbc.core.RowMapper; @@ -74,6 +74,62 @@ public class JdbcPollingChannelAdapterIntegrationTest { } + + @Test + public void testSimplePollForListWithRowMapperAndOneUpdate(){ + JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(this.embeddedDatabase, "select * from item where status=2"); + adapter.setUpdateQuery("update item set status = 10 where id in (:idList)"); + adapter.setRowMapper(new ItemRowMapper()); + + this.jdbcTemplate.update("insert into item values(1,2)") ; + this.jdbcTemplate.update("insert into item values(2,2)") ; + + Message message = adapter.receive(); + Object payload = message.getPayload(); + List rows = (List)payload; + assertEquals("Wrong number of elements" , 2, rows.size()); + assertTrue("Wrong payload type", rows.get(0) instanceof Item); + Item item = (Item)rows.get(0); + assertEquals("Wrong id", 1, item.getId()); + assertEquals("Wrong status", 2, item.getStatus()); + + int countOfStatusTwo = this.jdbcTemplate.queryForInt("select count(*) from item where status = 2"); + assertEquals("Status not updated incorect number of rows with status 2", 0, countOfStatusTwo); + + int countOfStatusTen = this.jdbcTemplate.queryForInt("select count(*) from item where status = 10"); + assertEquals("Status not updated incorect number of rows with status 10", 2, countOfStatusTen); + + + } + + @Test + public void testSimplePollForListWithRowMapperAndUpdatePerRow(){ + JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(this.embeddedDatabase, "select * from item where status=2"); + adapter.setUpdateQuery("update item set status = 10 where id = :id"); + adapter.setUpdatePerRow(true); + adapter.setRowMapper(new ItemRowMapper()); + + this.jdbcTemplate.update("insert into item values(1,2)") ; + this.jdbcTemplate.update("insert into item values(2,2)") ; + + Message message = adapter.receive(); + Object payload = message.getPayload(); + List rows = (List)payload; + assertEquals("Wrong number of elements" , 2, rows.size()); + assertTrue("Wrong payload type", rows.get(0) instanceof Item); + Item item = (Item)rows.get(0); + assertEquals("Wrong id", 1, item.getId()); + assertEquals("Wrong status", 2, item.getStatus()); + + int countOfStatusTwo = this.jdbcTemplate.queryForInt("select count(*) from item where status = 2"); + assertEquals("Status not updated incorect number of rows with status 2", 0, countOfStatusTwo); + + int countOfStatusTen = this.jdbcTemplate.queryForInt("select count(*) from item where status = 10"); + assertEquals("Status not updated incorect number of rows with status 10", 2, countOfStatusTen); + + + } + private static class Item{ @@ -114,4 +170,4 @@ public class JdbcPollingChannelAdapterIntegrationTest { } -} +} \ No newline at end of file