Resolves INT-790
This commit is contained in:
Jonas Partner
@@ -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<String, Object> staticParameters;
|
||||
|
||||
private volatile String polledRowIdName = "id";
|
||||
|
||||
private volatile String updateIdsParamName = "idList";
|
||||
|
||||
public DefaultSqlParamterSourceFactory() {
|
||||
this.staticParameters = Collections
|
||||
.unmodifiableMap(new HashMap<String, Object>());
|
||||
}
|
||||
|
||||
public DefaultSqlParamterSourceFactory(Map<String, Object> staticParameters) {
|
||||
this.staticParameters = Collections.unmodifiableMap(staticParameters);
|
||||
}
|
||||
|
||||
public SqlParameterSource createParamterSource(Object resultOfSelect) {
|
||||
SqlParameterSource toReturn;
|
||||
if (resultOfSelect instanceof List) {
|
||||
List<Object> ids = new ArrayList<Object>();
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Object>, InitializingBean {
|
||||
public class JdbcPollingChannelAdapter implements MessageSource<Object>,
|
||||
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<Object> receive() {
|
||||
Object payload = null;
|
||||
if (this.transactionTemplate != null){
|
||||
payload = this.transactionTemplate.execute(new TransactionCallback<Object>(){
|
||||
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<Object> receive() {
|
||||
Object payload = null;
|
||||
if (this.transactionTemplate != null) {
|
||||
payload = this.transactionTemplate
|
||||
.execute(new TransactionCallback<Object>() {
|
||||
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<Map<String, Object>> pollForListOfMap() {
|
||||
List<Map<String, Object>> payload = null;
|
||||
if (this.sqlQueryParameterSource != null) {
|
||||
payload = this.jdbcOperations.queryForList(this.selectQuery,
|
||||
this.sqlQueryParameterSource);
|
||||
} else {
|
||||
payload = this.jdbcOperations.queryForList(this.selectQuery);
|
||||
}
|
||||
return payload;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
}
|
||||
@@ -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<Object> 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<Object> 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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user