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

INT-791: outgoing JDBC adpater

Browse Source
This commit is contained in:
David Syer
2010-05-07 08:02:17 +00:00
parent 326a23d6fe
commit 4e8a5c2fb8
17 changed files with 530 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/DefaultSqlParameterSourceFactory.java
View File

@@ -1,17 +1,14 @@
/*
* Copyright 2002-2010 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.
*
* 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;
@@ -24,23 +21,20 @@ 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 SqlParameterSourceFactory} which creates an
* {@link SqlParameterSource} according to the result of the polled data passed in.
* A default implementation of {@link SqlParameterSourceFactory} which creates an {@link SqlParameterSource} according
* to the result of the data passed in.
*
* Where the result of the poll is a List, a list of ids is 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 data is a List, a list of ids is 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}.
* Where the data 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
* @since 2.0
@@ -55,7 +49,6 @@ public class DefaultSqlParameterSourceFactory implements SqlParameterSourceFacto
private final String updateIdsParamName = "idList";
public DefaultSqlParameterSourceFactory() {
this.staticParameters = Collections.unmodifiableMap(new HashMap<String, Object>());
}
@@ -64,7 +57,6 @@ public class DefaultSqlParameterSourceFactory implements SqlParameterSourceFacto
this.staticParameters = Collections.unmodifiableMap(staticParameters);
}
@SuppressWarnings("unchecked")
public SqlParameterSource createParameterSource(Object resultOfSelect) {
SqlParameterSource toReturn;
@@ -73,13 +65,11 @@ public class DefaultSqlParameterSourceFactory implements SqlParameterSourceFacto
for (Object rowObj : (List) resultOfSelect) {
if (rowObj instanceof Map) {
ids.add(((Map) rowObj).get(this.polledRowIdName));
}
else {
} else {
DirectFieldAccessor accessor = new DirectFieldAccessor(rowObj);
if (accessor.isReadableProperty(this.polledRowIdName)) {
ids.add(accessor.getPropertyValue(this.polledRowIdName));
}
else {
} else {
logger.warn("No id field named '" + this.polledRowIdName
+ "' found for result of polled row. Update may not include all rows.");
}
@@ -92,13 +82,11 @@ public class DefaultSqlParameterSourceFactory implements SqlParameterSourceFacto
thisParamSource.addValue(this.updateIdsParamName, ids);
thisParamSource.getValue("idList");
toReturn = thisParamSource;
}
else if (resultOfSelect instanceof Map) {
} else if (resultOfSelect instanceof Map) {
MapSqlParameterSource mapParameterSource = new MapSqlParameterSource((Map) resultOfSelect);
mapParameterSource.addValues(this.staticParameters);
toReturn = mapParameterSource;
}
else {
} else {
BeanPropertySqlParameterSource beanParameterSource = new BeanPropertySqlParameterSource(resultOfSelect);
toReturn = beanParameterSource;
}

101
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* Copyright 2002-2010 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 javax.sql.DataSource;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.jdbc.core.simple.SimpleJdbcOperations;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
/**
* A message handler that executes an SQL update. Dynamic query parameters are supported through the
* {@link SqlParameterSourceFactory} abstraction, the default implementation of which wraps the message so that its bean
* properties can be referred to by name in the query string E.g.
*
* <pre>
* INSERT INTO FOOS (MESSAGE_ID, PAYLOAD) VALUES (:headers[$id], :payload)
* </pre>
*
* N.B. do not use quotes to escape the header keys. The default SQL parameter source (from Spring JDBC) can also handle
* headers with dotted names (e.g. <code>business.id</code>)
*
* @author Dave Syer
* @since 2.0
*/
public class JdbcMessageHandler implements MessageHandler {
private final SimpleJdbcOperations jdbcOperations;
private volatile String updateSql;
private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new DefaultSqlParameterSourceFactory();
/**
* 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 updateSql query to execute
*/
public JdbcMessageHandler(DataSource dataSource, String updateSql) {
this.jdbcOperations = new SimpleJdbcTemplate(dataSource);
this.updateSql = updateSql;
}
/**
* Constructor taking {@link JdbcOperations} instance to use for query execution and the select query to execute to
* retrieve new rows.
*
* @param jdbcOperations instance to use for query execution
* @param updateSql query to execute
*/
public JdbcMessageHandler(JdbcOperations jdbcOperations, String updateSql) {
this.jdbcOperations = new SimpleJdbcTemplate(jdbcOperations);
this.updateSql = updateSql;
}
public void setUpdateSql(String updateSql) {
this.updateSql = updateSql;
}
public void setSqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) {
this.sqlParameterSourceFactory = sqlParameterSourceFactory;
}
/**
* Executes the update, passing the message into the {@link SqlParameterSourceFactory}.
*/
public void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
executeUpdateQuery(message);
}
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);
}
}
}

67
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* Copyright 2002-2010 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.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Dave Syer
* @since 2.0
*
*/
public class JdbcMessageHandlerParser extends AbstractOutboundChannelAdapterParser {
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
Object source = parserContext.extractSource(element);
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition("org.springframework.integration.jdbc.JdbcMessageHandler");
String dataSourceRef = element.getAttribute("data-source");
String jdbcOperationsRef = element.getAttribute("jdbc-operations");
boolean refToDataSourceSet = StringUtils.hasText(dataSourceRef);
boolean refToJdbcOperationsSet = StringUtils.hasText(jdbcOperationsRef);
if ((refToDataSourceSet && refToJdbcOperationsSet)
|| (!refToDataSourceSet && !refToJdbcOperationsSet)) {
parserContext.getReaderContext().error(
"Exactly one of the attributes data-source or "
+ "simple-jdbc-operations should be set for the JDBC outbound-channel-adapter", source);
}
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(jdbcOperationsRef);
}
builder.addConstructorArgValue(query);
return builder.getBeanDefinition();
}
}

1
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcNamespaceHandler.java
View File

@@ -29,6 +29,7 @@ public class JdbcNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new JdbcPollingChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new JdbcMessageHandlerParser());
registerBeanDefinitionParser("message-store", new JdbcMessageStoreParser());
}

10
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java
View File

@@ -47,11 +47,11 @@ public class JdbcPollingChannelAdapterParser extends AbstractPollingInboundChann
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition("org.springframework.integration.jdbc.JdbcPollingChannelAdapter");
String dataSourceRef = element.getAttribute("data-source");
String simpleJdbcOperationsRef = element.getAttribute("jdbc-operations");
String jdbcOperationsRef = element.getAttribute("jdbc-operations");
boolean refToDataSourceSet = StringUtils.hasText(dataSourceRef);
boolean refToSimpleJdbcOperaitonsSet = StringUtils.hasText(simpleJdbcOperationsRef);
if ((refToDataSourceSet && refToSimpleJdbcOperaitonsSet)
|| (!refToDataSourceSet && !refToSimpleJdbcOperaitonsSet)) {
boolean refToJdbcOperationsSet = StringUtils.hasText(jdbcOperationsRef);
if ((refToDataSourceSet && refToJdbcOperationsSet)
|| (!refToDataSourceSet && !refToJdbcOperationsSet)) {
parserContext.getReaderContext().error("Exactly one of the attributes data-source or " +
"simple-jdbc-operations should be set for the JDBC inbound-channel-adapter", source);
}
@@ -63,7 +63,7 @@ public class JdbcPollingChannelAdapterParser extends AbstractPollingInboundChann
builder.addConstructorArgReference(dataSourceRef);
}
else {
builder.addConstructorArgReference(simpleJdbcOperationsRef);
builder.addConstructorArgReference(jdbcOperationsRef);
}
builder.addConstructorArgValue(query);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper");

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

@@ -31,7 +31,7 @@
<xsd:annotation>
<xsd:documentation>
Reference to a data source to use to access
the database. Either this or the simple-jdbc-operations must be
the database. Either this or the jdbc-operations must be
specified (but not both).
</xsd:documentation>
<xsd:appinfo>
@@ -46,12 +46,13 @@
<xsd:appinfo>
<xsd:documentation>
Reference to a JdbcOperations. Either
this or the data-source must be
this or
the data-source must be
specified (but not both).
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.jdbc.core.simple.SimpleJdbcOperations" />
type="org.springframework.jdbc.core.JdbcOperations" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -108,21 +109,33 @@
<xsd:element ref="integration:poller" minOccurs="0"
maxOccurs="1" />
</xsd:sequence>
<xsd:attribute name="jdbc-operations" type="xsd:string">
<xsd:attribute name="data-source" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a data source to use to access
the
database. Either this or the simple-jdbc-operations must be
specified (but not both).
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.jdbc.core.simple.SimpleJdbcOperations" />
<tool:expected-type type="javax.sql.DataSource" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="data-source" type="xsd:string">
<xsd:attribute name="jdbc-operations" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Reference to a JdbcOperations. Either
this or
the data-source must be
specified (but not both).
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="javax.sql.DataSource" />
<tool:expected-type
type="org.springframework.jdbc.core.JdbcOperations" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -153,4 +166,67 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Channel Adapter for updating a
database.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="data-source" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a data source to use to access
the
database. Either this or the simple-jdbc-operations must be
specified (but not both).
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="javax.sql.DataSource" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="jdbc-operations" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Reference to a JdbcOperations. Either
this or
the data-source must be
specified (but not both).
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.jdbc.core.JdbcOperations" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="query" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
An SQL update query to execute (INSERT, UPDATE
or DELETE). Bean properties of the outgoing message can be
referenced in named parameters, e.g. "INSERT into FOOS (ID, NAME) values (:headers[business.key], :payload)"
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<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>

83
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java Normal file
View File

@@ -0,0 +1,83 @@
package org.springframework.integration.jdbc;
import static org.junit.Assert.assertEquals;
import java.util.Map;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.StringMessage;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
/**
* @author Dave Syer
*/
public class JdbcMessageHandlerIntegrationTests {
private EmbeddedDatabase embeddedDatabase;
private SimpleJdbcTemplate jdbcTemplate;
@Before
public void setUp() {
EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();
builder.setType(EmbeddedDatabaseType.HSQL).addScript(
"classpath:org/springframework/integration/jdbc/messageHandlerIntegrationTest.sql");
this.embeddedDatabase = builder.build();
this.jdbcTemplate = new SimpleJdbcTemplate(this.embeddedDatabase);
}
@After
public void tearDown() {
this.embeddedDatabase.shutdown();
}
@Test
public void testSimpleStaticInsert() {
JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate.getJdbcOperations(), "insert into foos (id, status, name) values (1, 0, 'foo')");
Message<String> message = new StringMessage("foo");
handler.handleMessage(message);
Map<String, Object> map = jdbcTemplate.queryForMap("SELECT * FROM FOOS WHERE ID=?", 1);
assertEquals("Wrong id", "1", map.get("ID"));
assertEquals("Wrong status", 0, map.get("STATUS"));
assertEquals("Wrong name", "foo", map.get("NAME"));
}
@Test
public void testSimpleDynamicInsert() {
JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate.getJdbcOperations(), "insert into foos (id, status, name) values (1, 0, :payload)");
Message<String> message = new StringMessage("foo");
handler.handleMessage(message);
Map<String, Object> map = jdbcTemplate.queryForMap("SELECT * FROM FOOS WHERE ID=?", 1);
assertEquals("Wrong name", "foo", map.get("NAME"));
}
@Test
public void testIdHeaderDynamicInsert() {
JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate.getJdbcOperations(), "insert into foos (id, status, name) values (:headers[$id], 0, :payload)");
Message<String> message = new StringMessage("foo");
handler.handleMessage(message);
String id = message.getHeaders().getId().toString();
Map<String, Object> map = jdbcTemplate.queryForMap("SELECT * FROM FOOS WHERE ID=?", id);
assertEquals("Wrong id", id, map.get("ID"));
assertEquals("Wrong name", "foo", map.get("NAME"));
}
@Test
public void testDottedHeaderDynamicInsert() {
JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate.getJdbcOperations(), "insert into foos (id, status, name) values (:headers[business.id], 0, :payload)");
Message<String> message = MessageBuilder.withPayload("foo").setHeader("business.id", "FOO").build();
handler.handleMessage(message);
String id = message.getHeaders().get("business.id").toString();
Map<String, Object> map = jdbcTemplate.queryForMap("SELECT * FROM FOOS WHERE ID=?", id);
assertEquals("Wrong id", id, map.get("ID"));
assertEquals("Wrong name", "foo", map.get("NAME"));
}
}

25
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java → org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java
View File

@@ -1,6 +1,8 @@
package org.springframework.integration.jdbc;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -13,7 +15,6 @@ 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;
@@ -21,11 +22,11 @@ import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
/**
* @author Jonas Partner
*/
public class JdbcPollingChannelAdapterIntegrationTest {
public class JdbcPollingChannelAdapterIntegrationTests {
EmbeddedDatabase embeddedDatabase;
private EmbeddedDatabase embeddedDatabase;
SimpleJdbcTemplate jdbcTemplate;
private SimpleJdbcTemplate jdbcTemplate;
@Before
@@ -51,11 +52,11 @@ public class JdbcPollingChannelAdapterIntegrationTest {
this.jdbcTemplate.update("insert into item values(1,2)");
Message<Object> message = adapter.receive();
Object payload = message.getPayload();
assertTrue("Wrong payload type", payload instanceof List);
List rows = (List) payload;
assertTrue("Wrong payload type", payload instanceof List<?>);
List<?> rows = (List<?>) payload;
assertEquals("Wrong number of elements", 1, rows.size());
assertTrue("Returned row not a map", rows.get(0) instanceof Map);
Map<String, Object> row = (Map<String, Object>) rows.get(0);
assertTrue("Returned row not a map", rows.get(0) instanceof Map<?,?>);
Map<?, ?> row = (Map<?, ?>) rows.get(0);
assertEquals("Wrong id", 1, row.get("id"));
assertEquals("Wrong status", 2, row.get("status"));
@@ -69,7 +70,7 @@ public class JdbcPollingChannelAdapterIntegrationTest {
this.jdbcTemplate.update("insert into item values(1,2)");
Message<Object> message = adapter.receive();
Object payload = message.getPayload();
List rows = (List) payload;
List<?> rows = (List<?>) payload;
assertEquals("Wrong number of elements", 1, rows.size());
assertTrue("Wrong payload type", rows.get(0) instanceof Item);
Item item = (Item) rows.get(0);
@@ -91,7 +92,7 @@ public class JdbcPollingChannelAdapterIntegrationTest {
Message<Object> message = adapter.receive();
Object payload = message.getPayload();
List rows = (List) payload;
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);
@@ -125,7 +126,7 @@ public class JdbcPollingChannelAdapterIntegrationTest {
Message<Object> message = adapter.receive();
Object payload = message.getPayload();
List rows = (List) payload;
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);

70
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java Normal file
View File

@@ -0,0 +1,70 @@
package org.springframework.integration.jdbc.config;
import static org.junit.Assert.assertEquals;
import java.util.Collections;
import java.util.Map;
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.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
public class JdbcMessageHandlerParserTests {
private SimpleJdbcTemplate jdbcTemplate;
private MessageChannel channel;
private ConfigurableApplicationContext context;
@Test
public void testSimpleInboundChannelAdapter(){
setUp("handlingWithJdbcOperationsJdbcOutboundChannelAdapterTest.xml", getClass());
Message<?> message = MessageBuilder.withPayload("foo").setHeader("business.key", "FOO").build();
channel.send(message);
Map<String, Object> map = this.jdbcTemplate.queryForMap("SELECT * from FOOS");
assertEquals("Wrong id", "FOO", map.get("ID"));
assertEquals("Wrong id", "foo", map.get("name"));
}
@Test
public void testDollarHeaderInboundChannelAdapter(){
setUp("handlingDollarHeaderJdbcOutboundChannelAdapterTest.xml", getClass());
Message<?> message = MessageBuilder.withPayload("foo").build();
channel.send(message);
Map<String, Object> map = this.jdbcTemplate.queryForMap("SELECT * from FOOS");
assertEquals("Wrong id", message.getHeaders().getId().toString(), map.get("ID"));
assertEquals("Wrong id", "foo", map.get("name"));
}
@Test
public void testMapPayloadInboundChannelAdapter(){
setUp("handlingMapPayloadJdbcOutboundChannelAdapterTest.xml", getClass());
Message<?> message = MessageBuilder.withPayload(Collections.singletonMap("foo", "bar")).build();
channel.send(message);
Map<String, Object> map = this.jdbcTemplate.queryForMap("SELECT * from FOOS");
assertEquals("Wrong id", message.getHeaders().getId().toString(), map.get("ID"));
assertEquals("Wrong id", "bar", map.get("name"));
}
@After
public void tearDown(){
if(context != null){
context.close();
}
}
public void setUp(String name, Class<?> cls){
context = new ClassPathXmlApplicationContext(name, cls);
jdbcTemplate = new SimpleJdbcTemplate(this.context.getBean("dataSource",DataSource.class));
channel = this.context.getBean("target", MessageChannel.class);
}
}

77
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTest.java
View File

@@ -1,77 +0,0 @@
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;
@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));
}
}

17
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/handlingDollarHeaderJdbcOutboundChannelAdapterTest.xml Normal file
View File

@@ -0,0 +1,17 @@
<?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/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.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">
<outbound-channel-adapter query="insert into foos (id, status, name) values (:headers[$id], 0, :payload)"
channel="target" data-source="dataSource"/>
<beans:import resource="jdbcOutboundChannelAdapterCommonConfig.xml" />
</beans:beans>

17
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/handlingMapPayloadJdbcOutboundChannelAdapterTest.xml Normal file
View File

@@ -0,0 +1,17 @@
<?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/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.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">
<outbound-channel-adapter query="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])"
channel="target" data-source="dataSource"/>
<beans:import resource="jdbcOutboundChannelAdapterCommonConfig.xml" />
</beans:beans>

17
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/handlingWithJdbcOperationsJdbcOutboundChannelAdapterTest.xml Normal file
View File

@@ -0,0 +1,17 @@
<?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/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.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">
<outbound-channel-adapter query="insert into foos (id, status, name) values (:headers[business.key], 0, :payload)"
channel="target" jdbc-operations="jdbcTemplate" />
<beans:import resource="jdbcOutboundChannelAdapterCommonConfig.xml" />
</beans:beans>

22
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml
View File

@@ -1,9 +1,7 @@
<?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"
<beans xmlns="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"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.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">
@@ -13,15 +11,15 @@
</si:channel>
<jdbc:embedded-database type="HSQL" id="dataSource">
<jdbc:script location="org/springframework/integration/jdbc/config/inboundSchema.sql" />
<jdbc:script location="org/springframework/integration/jdbc/config/outboundSchema.sql" />
</jdbc:embedded-database>
<beans:bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<beans:property name="dataSource" ref="dataSource"/>
</beans:bean>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<si:poller default="true">
<si:interval-trigger interval="100" />
</si:poller>
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<constructor-arg ref="dataSource" />
</bean>
</beans:beans>
</beans>

24
org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/config/jdbcOutboundChannelAdapterCommonConfig.xml Normal file
View File

@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:si="http://www.springframework.org/schema/integration"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.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"/>
<jdbc:embedded-database type="HSQL" id="dataSource">
<jdbc:script location="org/springframework/integration/jdbc/config/outboundSchema.sql" />
</jdbc:embedded-database>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<constructor-arg ref="dataSource" />
</bean>
</beans>

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

@@ -0,0 +1 @@
create table foos(id varchar(100),status int,name varchar(20));

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

@@ -0,0 +1 @@
create table foos(id varchar(100),status int,name varchar(20));