From 79b72a37949531801774764eabcbd8c6652f0fb5 Mon Sep 17 00:00:00 2001 From: David Syer Date: Sat, 10 Jul 2010 16:44:38 +0000 Subject: [PATCH] INT-1231: add support for max-rows-per-poll --- .../jdbc/JdbcPollingChannelAdapter.java | 96 ++++++++++++------- .../JdbcPollingChannelAdapterParser.java | 1 + .../config/spring-integration-jdbc-2.0.xsd | 10 ++ .../JdbcPollingChannelAdapterParserTests.java | 30 ++++++ .../jdbcInboundChannelAdapterCommonConfig.xml | 6 +- ...thMaxRowsJdbcInboundChannelAdapterTest.xml | 18 ++++ 6 files changed, 128 insertions(+), 33 deletions(-) create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml 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 57158d73a6..5cfbb090b6 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 @@ -16,15 +16,22 @@ package org.springframework.integration.jdbc; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; import java.util.List; import javax.sql.DataSource; +import org.springframework.dao.DataAccessException; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageSource; +import org.springframework.jdbc.core.ColumnMapRowMapper; import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.ResultSetExtractor; import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.RowMapperResultSetExtractor; import org.springframework.jdbc.core.namedparam.SqlParameterSource; import org.springframework.jdbc.core.simple.SimpleJdbcOperations; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; @@ -35,6 +42,7 @@ import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; * select in order to update processed rows. * * @author Jonas Partner + * @author Dave Syer * @since 2.0 */ public class JdbcPollingChannelAdapter implements MessageSource { @@ -53,10 +61,11 @@ public class JdbcPollingChannelAdapter implements MessageSource { private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new DefaultSqlParameterSourceFactory(); + private int maxRowsPerPoll = 0; /** - * Constructor taking {@link DataSource} from which the DB Connection can - * be obtained and the select query to execute to retrieve new rows. + * Constructor taking {@link DataSource} from which the DB Connection can be + * obtained and the select query to execute to retrieve new rows. * * @param dataSource used to create a {@link SimpleJdbcTemplate} * @param selectQuery query to execute @@ -78,7 +87,6 @@ public class JdbcPollingChannelAdapter implements MessageSource { this.selectQuery = selectQuery; } - public void setRowMapper(RowMapper rowMapper) { this.rowMapper = rowMapper; } @@ -94,22 +102,32 @@ public class JdbcPollingChannelAdapter implements MessageSource { 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) { + public void setSqlQueryParameterSource(SqlParameterSource sqlQueryParameterSource) { this.sqlQueryParameterSource = sqlQueryParameterSource; } /** - * Executes the query. If a query result set contains one or more rows, the Message - * payload will contain either a List of Maps for each row or, if a RowMapper has - * been provided, the values mapped from those rows. If the query returns no rows, - * this method will return null. + * The maximum number of rows to pull out of the query results per poll (if + * greater than zero, otherwise all rows will be packed into the outgoing + * message). Default is zero. + * + * @param maxRows the max rows to set + */ + public void setMaxRowsPerPoll(int maxRows) { + this.maxRowsPerPoll = maxRows; + } + + /** + * Executes the query. If a query result set contains one or more rows, the + * Message payload will contain either a List of Maps for each row or, if a + * RowMapper has been provided, the values mapped from those rows. If the + * query returns no rows, this method will return null. */ public Message receive() { Object payload = poll(); @@ -120,18 +138,12 @@ public class JdbcPollingChannelAdapter implements MessageSource { } /** - * Execute the select query and the update query if provided. - * Returns the rows returned by the select query. If a RowMapper - * has been provided, the mapped results are returned. + * Execute the select query and the update query if provided. Returns the + * rows returned by the select query. If a RowMapper has been provided, the + * mapped results are returned. */ private Object poll() { - List payload; - if (this.rowMapper != null) { - payload = pollWithRowMapper(); - } - else { - payload = this.jdbcOperations.queryForList(this.selectQuery, this.sqlQueryParameterSource); - } + List payload = doPoll(); if (payload.size() < 1) { payload = null; } @@ -149,25 +161,45 @@ public class JdbcPollingChannelAdapter implements MessageSource { } private void executeUpdateQuery(Object obj) { - SqlParameterSource updateParamaterSource = null; - if (this.sqlParameterSourceFactory != null) { - updateParamaterSource = this.sqlParameterSourceFactory.createParameterSource(obj); - this.jdbcOperations.update(this.updateSql, updateParamaterSource); - } - else { - this.jdbcOperations.update(this.updateSql); - } + SqlParameterSource updateParamaterSource = this.sqlParameterSourceFactory.createParameterSource(obj); + this.jdbcOperations.update(this.updateSql, updateParamaterSource); } - private List pollWithRowMapper() { + private List doPoll() { + List payload = null; - if (this.sqlQueryParameterSource != null) { - payload = this.jdbcOperations.query(this.selectQuery, this.rowMapper, this.sqlQueryParameterSource); + final RowMapper rowMapper = this.rowMapper == null ? new ColumnMapRowMapper() : this.rowMapper; + ResultSetExtractor> resultSetExtractor; + + if (maxRowsPerPoll > 0) { + resultSetExtractor = new ResultSetExtractor>() { + public List extractData(ResultSet rs) throws SQLException, DataAccessException { + List results = new ArrayList(maxRowsPerPoll); + int rowNum = 0; + while (rs.next() && rowNum < maxRowsPerPoll) { + results.add(rowMapper.mapRow(rs, rowNum++)); + } + return results; + } + }; } else { - payload = this.jdbcOperations.query(this.selectQuery, this.rowMapper); + @SuppressWarnings("unchecked") + ResultSetExtractor> temp = new RowMapperResultSetExtractor( + (RowMapper) rowMapper); + resultSetExtractor = temp; } + + if (this.sqlQueryParameterSource != null) { + payload = this.jdbcOperations.getNamedParameterJdbcOperations().query(this.selectQuery, + this.sqlQueryParameterSource, resultSetExtractor); + } + else { + payload = this.jdbcOperations.getJdbcOperations().query(this.selectQuery, resultSetExtractor); + } + return payload; + } } 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 0ed9ef43dd..e324353693 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 @@ -70,6 +70,7 @@ public class JdbcPollingChannelAdapterParser extends AbstractPollingInboundChann IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-parameter-source-factory"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-query-parameter-source"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-rows-per-poll"); if (update!=null) { builder.addPropertyValue("updateSql", update); } 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 f2a4269c84..bf79702f87 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 @@ -202,6 +202,16 @@ + + + + + Limits the number of rows extracted per query (otherwise all rows + are extracted into the outgoing message). + + + + 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 4ec8d811f9..aa631d8fca 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 @@ -19,7 +19,11 @@ 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.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; @Transactional public class JdbcPollingChannelAdapterParserTests { @@ -32,6 +36,8 @@ public class JdbcPollingChannelAdapterParserTests { private MessageChannelTemplate channelTemplate; private ConfigurableApplicationContext appCtx; + + private PlatformTransactionManager transactionManager; @Test public void testSimpleInboundChannelAdapter(){ @@ -90,6 +96,24 @@ public class JdbcPollingChannelAdapterParserTests { assertNotNull(message); } + @Test + public void testMaxRowsInboundChannelAdapter(){ + setUp("pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml", getClass()); + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Void doInTransaction(TransactionStatus status) { + jdbcTemplate.update("insert into item values(1,'',2)"); + jdbcTemplate.update("insert into item values(2,'',2)"); + jdbcTemplate.update("insert into item values(3,'',2)"); + jdbcTemplate.update("insert into item values(4,'',2)"); + return null; + } + }); + @SuppressWarnings("unchecked") + Message> message = (Message>) channelTemplate.receive(); + assertNotNull(message); + assertEquals(2, message.getPayload().size()); + } + @After public void tearDown(){ if(appCtx != null){ @@ -100,6 +124,8 @@ public class JdbcPollingChannelAdapterParserTests { public void setUp(String name, Class cls){ appCtx = new ClassPathXmlApplicationContext(name, cls); setupJdbcTemplate(); + jdbcTemplate.update("delete from item"); + setupTransactionManager(); setupMessageChannelTemplate(); } @@ -114,6 +140,10 @@ public class JdbcPollingChannelAdapterParserTests { this.jdbcTemplate = new SimpleJdbcTemplate(this.appCtx.getBean("dataSource",DataSource.class)); } + protected void setupTransactionManager(){ + this.transactionManager = this.appCtx.getBean("transactionManager",PlatformTransactionManager.class); + } + public static class TestSqlParameterSource extends AbstractSqlParameterSource { public Object getValue(String paramName) diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml index 5b4080cab2..0999e6fba8 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml @@ -10,7 +10,7 @@ - + @@ -18,6 +18,10 @@ + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml new file mode 100644 index 0000000000..2ddf63cef3 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml @@ -0,0 +1,18 @@ + + + + + + + +