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

namespace support for JDBC inbound channel adapter

Browse Source 
resolves INT-876
This commit is contained in:
Jonas Partner
2010-03-10 14:26:40 +00:00
parent 1b2ff13413
commit 304f4f5845
13 changed files with 381 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java
View File

@@ -17,11 +17,9 @@
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;
import org.springframework.integration.message.MessageSource;
@@ -29,12 +27,6 @@ import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.jdbc.core.simple.SimpleJdbcOperations;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
* A polling channel adapter that creates messages from the payload returned by
@@ -43,8 +35,7 @@ import org.springframework.transaction.support.TransactionTemplate;
*
* @author Jonas Partner
*/
public class JdbcPollingChannelAdapter implements MessageSource<Object>,
InitializingBean {
public class JdbcPollingChannelAdapter implements MessageSource<Object> {
private final SimpleJdbcOperations jdbcOperations;
@@ -54,48 +45,49 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object>,
private volatile SqlParameterSource sqlQueryParameterSource;
private volatile TransactionDefinition transactionDefinition;
private volatile TransactionTemplate transactionTemplate;
private volatile PlatformTransactionManager platformTransactionManager;
private volatile boolean updatePerRow = false;
private volatile String updateSql;
private volatile SqlParamterSourceFactory sqlParameterSourceFactoryForUpdate = new DefaultSqlParamterSourceFactory();
/**
* Constructor taking query to execute to retreive new rows and
* {@link DataSource} from which the DB Connection can be obtained
*
* @param dataSource
* used to create a {@link SimpleJdbcTemplate}
* @param selectQuery
* query to execute
*/
public JdbcPollingChannelAdapter(DataSource dataSource, String selectQuery) {
this.jdbcOperations = new SimpleJdbcTemplate(dataSource);
this.selectQuery = selectQuery;
}
/**
* Constructor taking query to execute on a poll and
* {@link SimpleJdbcOperations} instance to use for query execution
*
* @param jdbcOperations
* @param selectQuery
* query to execute
*/
public JdbcPollingChannelAdapter(SimpleJdbcOperations jdbcOperations,
String selectQuery) {
this.jdbcOperations = jdbcOperations;
this.selectQuery = selectQuery;
}
public void setTransactionDefinition(
TransactionDefinition transactionDefinition) {
this.transactionDefinition = transactionDefinition;
}
public void setTransactionManager(
PlatformTransactionManager platformTransactionManager) {
this.platformTransactionManager = platformTransactionManager;
}
public void setRowMapper(RowMapper<?> rowMapper) {
this.rowMapper = rowMapper;
}
public void setUpdatesql(String updateSql) {
public void setUpdateSql(String updateSql) {
this.updateSql = updateSql;
}
public void setUpdatePerRow(boolean updatePerRow){
public void setUpdatePerRow(boolean updatePerRow) {
this.updatePerRow = updatePerRow;
}
@@ -104,35 +96,25 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object>,
this.sqlParameterSourceFactoryForUpdate = sqlParameterSourceFactoryForUpdate;
}
public void afterPropertiesSet() throws Exception {
if (this.transactionDefinition == null) {
this.transactionDefinition = new DefaultTransactionDefinition();
}
if (this.platformTransactionManager != null) {
this.transactionTemplate = new TransactionTemplate(
this.platformTransactionManager, this.transactionDefinition);
}
}
/**
* Polls for new rows returning a message containing one or more rows where
* rows are found and null where no rows are returned by the select query
*/
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();
}
if(payload == null){
payload = pollAndUpdate();
if (payload == null) {
return null;
}
return MessageBuilder.withPayload(payload).build();
}
/**
* Execute the select query and the update query if provided and rows are
* returned by the select query
*
* @return
*/
protected Object pollAndUpdate() {
List payload;
if (this.rowMapper != null) {
@@ -141,11 +123,11 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object>,
payload = this.jdbcOperations.queryForList(this.selectQuery,
this.sqlQueryParameterSource);
}
if(payload.size() < 1){
if (payload.size() < 1) {
payload = null;
}
if (payload != null && updateSql != null) {
if (this.updatePerRow) {
for (Object row : payload) {
@@ -162,7 +144,7 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object>,
protected void executeUpdateQuery(Object obj) {
SqlParameterSource updateParamaterSource = null;
if (this.sqlParameterSourceFactoryForUpdate != null) {
updateParamaterSource = this.sqlParameterSourceFactoryForUpdate
.createParamterSource(obj);
this.jdbcOperations.update(this.updateSql, updateParamaterSource);
@@ -183,15 +165,4 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object>,
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;
}
}

31
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcNamespaceHandler.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2002-2009 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.config;
import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
/**
*
* @author Jonas Partner
*
*/
public class JdbcNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new JdbcPollingChannelAdapterParser());
}
}

83
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2002-2009 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.config;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for {@link JdbcPollingChannelAdapterParser}
* @author Jonas Partner
*
*/
public class JdbcPollingChannelAdapterParser extends
AbstractPollingInboundChannelAdapterParser {
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected String parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition("org.springframework.integration.jdbc.JdbcPollingChannelAdapter");
String dataSourceRef = element.getAttribute("data-source");
String simpleJdbcOperationsRef = element
.getAttribute("simple-jdbc-operations");
boolean refToDataSourceSet = StringUtils.hasText(dataSourceRef);
boolean refToSimpleJdbcOperaitonsSet = StringUtils
.hasText(simpleJdbcOperationsRef);
if ((refToDataSourceSet && refToSimpleJdbcOperaitonsSet)
|| (!refToDataSourceSet && !refToSimpleJdbcOperaitonsSet)) {
throw new BeanCreationException(
"Exactly one of the attributes data-source or simple-jdbc-operations should be set for the JDBC inbound-channel-adapter");
}
String query = element.getAttribute("query");
if(!StringUtils.hasText(query)){
throw new BeanCreationException("The query attrbitue is required");
}
if(refToDataSourceSet){
builder.addConstructorArgReference(dataSourceRef);
} else {
builder.addConstructorArgReference(simpleJdbcOperationsRef);
}
builder.addConstructorArgValue(query);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "update", "updateSql");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "update-per-row");
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder
.getBeanDefinition(), parserContext.getRegistry());
}
}

1
org.springframework.integration.jdbc/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/jdbc=org.springframework.integration.jdbc.config.JdbcNamespaceHandler

2
org.springframework.integration.jdbc/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc-2.0.xsd=org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd
http\://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd=org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd

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

@@ -0,0 +1,77 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/jdbc"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/jdbc"
elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans" />
<xsd:import namespace="http://www.springframework.org/schema/tool" />
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd" />
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration's JDBC adapters.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an inbound Channel Adapter for polling a database.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0"
maxOccurs="1" />
</xsd:sequence>
<xsd:attribute name="simple-jdbc-operations" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="import org.springframework.jdbc.core.simple.SimpleJdbcOperations" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="data-source" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="import javax.sql.DataSource" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="query" type="xsd:string" use="required" />
<xsd:attribute name="row-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="import org.springframework.jdbc.core.RowMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="update" type="xsd:string" />
<xsd:attribute name="update-per-row" type="xsd:boolean"
default="false" />
<xsd:attribute name="channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>

0
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/config/spring-integration-jdbc.gif → org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc.gif
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 572 B

After

Width:  |  Height:  |  Size: 572 B

8
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java
View File

@@ -13,6 +13,7 @@ import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
@@ -26,6 +27,7 @@ public class JdbcPollingChannelAdapterIntegrationTest {
SimpleJdbcTemplate jdbcTemplate;
@Before
public void setUp() {
EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();
@@ -59,8 +61,6 @@ public class JdbcPollingChannelAdapterIntegrationTest {
}
@Test
public void testSimplePollForListWithRowMapperNoUpdate() {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
@@ -83,7 +83,7 @@ public class JdbcPollingChannelAdapterIntegrationTest {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
this.embeddedDatabase, "select * from item where status=2");
adapter
.setUpdatesql("update item set status = 10 where id in (:idList)");
.setUpdateSql("update item set status = 10 where id in (:idList)");
adapter.setRowMapper(new ItemRowMapper());
this.jdbcTemplate.update("insert into item values(1,2)");
@@ -116,7 +116,7 @@ public class JdbcPollingChannelAdapterIntegrationTest {
public void testSimplePollForListWithRowMapperAndUpdatePerRow() {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
this.embeddedDatabase, "select * from item where status=2");
adapter.setUpdatesql("update item set status = 10 where id = :id");
adapter.setUpdateSql("update item set status = 10 where id = :id");
adapter.setUpdatePerRow(true);
adapter.setRowMapper(new ItemRowMapper());

79
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTest.java Normal file
View File

@@ -0,0 +1,79 @@
package org.springframework.integration.jdbc.config;
import static org.junit.Assert.*;
import java.util.List;
import javax.sql.DataSource;
import org.junit.After;
import org.junit.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.transaction.annotation.Transactional;
import com.sun.xml.internal.xsom.impl.scd.Iterators.Map;
@Transactional
public class JdbcPollingChannelAdapterParserTest {
final long receiveTimeout = 5000;
SimpleJdbcTemplate jdbcTemplate;
MessageChannelTemplate channelTemplate;
ConfigurableApplicationContext appCtx;
@Test
public void testSimpleInboundChannelAdapter(){
setUp("pollingForMapJdbcInboundChannelAdapterTest.xml", getClass());
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);
}
@Test
public void testSimpleInboundChannelAdapterWithUpdate(){
setUp("pollingForMapJdbcInboundChannelAdapterWithUpdateTest.xml", getClass());
this.jdbcTemplate.update("insert into item values(1,2)");
Message<?> message = channelTemplate.receive();
assertNotNull(message);
message = channelTemplate.receive();
assertNull(channelTemplate.receive());
}
@After
public void tearDown(){
if(appCtx != null){
appCtx.close();
}
}
public void setUp(String name, Class<?> cls){
appCtx = new ClassPathXmlApplicationContext(name, cls);
setupJdbcTemplate();
setupMessageChannelTemplate();
}
protected void setupMessageChannelTemplate(){
PollableChannel pollableChannel = this.appCtx.getBean("target", PollableChannel.class);
this.channelTemplate = new MessageChannelTemplate(pollableChannel);
this.channelTemplate.setReceiveTimeout(5000);
}
protected void setupJdbcTemplate(){
this.jdbcTemplate = new SimpleJdbcTemplate(this.appCtx.getBean("dataSource",DataSource.class));
}
}

1
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/inboundSchema.sql Normal file
View File

@@ -0,0 +1 @@
create table item(id int,status int);

26
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml Normal file
View File

@@ -0,0 +1,26 @@
<?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" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<si:channel id="target">
<si:queue />
</si:channel>
<jdbc:embedded-database type="DERBY" id="dataSource">
<jdbc:script location="org/springframework/integration/jdbc/config/inboundSchema.sql" />
</jdbc:embedded-database>
<beans:bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager" >
<beans:property name="dataSource" ref="dataSource"/>
</beans:bean>
<si:poller default="true">
<si:interval-trigger interval="1" />
</si:poller>
</beans:beans>

18
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/pollingForMapJdbcInboundChannelAdapterTest.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" />
<beans:import resource="jdbcInboundChannelAdapterCommonConfig.xml" />
</beans:beans>

22
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/pollingForMapJdbcInboundChannelAdapterWithUpdateTest.xml Normal file
View File

@@ -0,0 +1,22 @@
<?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" update="update item set status =10 where id in (:idList)" />
<beans:import resource="jdbcInboundChannelAdapterCommonConfig.xml" />
</beans:beans>