Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1231: add support for max-rows-per-poll

Browse Source
This commit is contained in:
David Syer
2010-07-10 16:44:38 +00:00
parent dd102fa2e0
commit 79b72a3794
6 changed files with 128 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Object> {
@@ -53,10 +61,11 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
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<Object> {
this.selectQuery = selectQuery;
}
public void setRowMapper(RowMapper<?> rowMapper) {
this.rowMapper = rowMapper;
}
@@ -94,22 +102,32 @@ 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) {
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 <code>null</code>.
* 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 <code>null</code>.
*/
public Message<Object> receive() {
Object payload = poll();
@@ -120,18 +138,12 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
}
/**
* 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<Object> {
}
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<List<Object>> resultSetExtractor;
if (maxRowsPerPoll > 0) {
resultSetExtractor = new ResultSetExtractor<List<Object>>() {
public List<Object> extractData(ResultSet rs) throws SQLException, DataAccessException {
List<Object> results = new ArrayList<Object>(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<List<Object>> temp = new RowMapperResultSetExtractor<Object>(
(RowMapper<Object>) 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;
}
}

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

@@ -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);
}

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

@@ -202,6 +202,16 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="max-rows-per-poll" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Limits the number of rows extracted per query (otherwise all rows
are extracted into the outgoing message).
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>

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

@@ -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<Void>() {
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<List<?>> message = (Message<List<?>>) 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)

6
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml
View File

@@ -10,7 +10,7 @@
<si:queue />
</si:channel>
<si:poller default="true">
<si:poller default="true" task-executor="taskExecutor">
<si:interval-trigger interval="100" />
</si:poller>
@@ -18,6 +18,10 @@
<jdbc:script location="org/springframework/integration/jdbc/config/inboundSchema.sql" />
</jdbc:embedded-database>
<bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor">
<property name="concurrencyLimit" value="1"/>
</bean>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>

18
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml Normal file
View File

@@ -0,0 +1,18 @@
<?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=2"
channel="target" data-source="dataSource" max-rows-per-poll="2"
update="update item set status=10 where id in (:idList)" />
<beans:import resource="jdbcInboundChannelAdapterCommonConfig.xml" />
</beans:beans>