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

INT-1327: add support for JDBC outbound gateway

Browse Source
This commit is contained in:
David Syer
2010-09-01 14:09:09 +00:00
parent b654e566eb
commit d2a3c3d5eb
22 changed files with 782 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java
View File

@@ -28,12 +28,13 @@ import org.springframework.jdbc.core.namedparam.SqlParameterSource;
/**
* An implementation of {@link SqlParameterSourceFactory} which creates an {@link SqlParameterSource} that evaluates
* Spring EL expressions. In addition the user can supply static parameters that always take precedence.
* Spring EL expressions. In addition the user can supply static parameters that always take precedence.
*
* @author Dave Syer
* @since 2.0
*/
public class ExpressionEvaluatingSqlParameterSourceFactory extends AbstractExpressionEvaluator implements SqlParameterSourceFactory {
public class ExpressionEvaluatingSqlParameterSourceFactory extends AbstractExpressionEvaluator implements
SqlParameterSourceFactory {
private final static Log logger = LogFactory.getLog(ExpressionEvaluatingSqlParameterSourceFactory.class);
@@ -77,20 +78,24 @@ public class ExpressionEvaluatingSqlParameterSourceFactory extends AbstractExpre
}
String expression = paramName;
if (input instanceof Collection<?>) {
expression = "#root.!["+paramName+"]";
expression = "#root.![" + paramName + "]";
}
Object value = evaluateExpression(expression, input);
values.put(paramName, value);
if (logger.isDebugEnabled()) {
logger.debug("Resolved expression " + expression + " to " + value);
}
return value;
}
public boolean hasValue(String paramName) {
try {
Object value = getValue(paramName);
if (value==ERROR) {
return false;
if (value == ERROR) {
return false;
}
} catch (ExpressionException e) {
}
catch (ExpressionException e) {
if (logger.isDebugEnabled()) {
logger.debug("Could not evaluate expression", e);
}

51
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java
View File

@@ -13,6 +13,10 @@
package org.springframework.integration.jdbc;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;
import org.springframework.integration.Message;
@@ -21,9 +25,13 @@ import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.jdbc.core.simple.SimpleJdbcOperations;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.jdbc.support.GeneratedKeyHolder;
import org.springframework.jdbc.support.KeyHolder;
import org.springframework.util.LinkedCaseInsensitiveMap;
/**
* A message handler that executes an SQL update. Dynamic query parameters are supported through the
@@ -48,6 +56,8 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new BeanPropertySqlParameterSourceFactory();
private boolean keysGenerated;
/**
* Constructor taking {@link DataSource} from which the DB Connection can be obtained and the select query to
* execute to retrieve new rows.
@@ -72,6 +82,14 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
this.updateSql = updateSql;
}
/**
* Flag to indicate that the update query is an insert with autogenerated keys, which will be logged at debug level.
* @param keysGenerated the flag value to set
*/
public void setKeysGenerated(boolean keysGenerated) {
this.keysGenerated = keysGenerated;
}
public void setUpdateSql(String updateSql) {
this.updateSql = updateSql;
}
@@ -85,17 +103,30 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
*/
protected void handleMessageInternal(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);
List<? extends Map<String, Object>> keys = executeUpdateQuery(message, keysGenerated);
if (logger.isDebugEnabled() && !keys.isEmpty()) {
logger.debug("Generated keys: "+keys);
}
}
protected List<? extends Map<String, Object>> executeUpdateQuery(Object obj, boolean keysGenerated) {
SqlParameterSource updateParameterSource = new MapSqlParameterSource();
if (this.sqlParameterSourceFactory != null) {
updateParameterSource = this.sqlParameterSourceFactory.createParameterSource(obj);
}
if (keysGenerated) {
KeyHolder keyHolder = new GeneratedKeyHolder();
this.jdbcOperations.getNamedParameterJdbcOperations().update(this.updateSql, updateParameterSource,
keyHolder);
return keyHolder.getKeyList();
}
else {
int updated = this.jdbcOperations.update(this.updateSql, updateParameterSource);
LinkedCaseInsensitiveMap<Object> map = new LinkedCaseInsensitiveMap<Object>();
map.put("UPDATED", updated);
return Collections.singletonList(map);
}
}
}

130
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java Normal file
View File

@@ -0,0 +1,130 @@
/*
* 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 java.util.List;
import javax.sql.DataSource;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
/**
* @author Dave Syer
*
* @since 2.0
*/
public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler implements InitializingBean {
private final JdbcMessageHandler handler;
private final JdbcPollingChannelAdapter poller;
private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new ExpressionEvaluatingSqlParameterSourceFactory();
private volatile boolean keysGenerated;
public JdbcOutboundGateway(DataSource dataSource, String updateQuery) {
this(new JdbcTemplate(dataSource), updateQuery, null);
}
public JdbcOutboundGateway(DataSource dataSource, String updateQuery, String selectQuery) {
this(new JdbcTemplate(dataSource), updateQuery, selectQuery);
}
public JdbcOutboundGateway(JdbcOperations jdbcOperations, String updateQuery) {
this(jdbcOperations, updateQuery, null);
}
public JdbcOutboundGateway(JdbcOperations jdbcOperations, String updateQuery, String selectQuery) {
if (selectQuery != null) {
poller = new JdbcPollingChannelAdapter(jdbcOperations, selectQuery);
poller.setMaxRowsPerPoll(1);
}
else {
poller = null;
}
handler = new JdbcMessageHandler(jdbcOperations, updateQuery);
}
public void setMaxRowsPerPoll(int maxRows) {
poller.setMaxRowsPerPoll(maxRows);
}
@Override
protected void onInit() {
handler.afterPropertiesSet();
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
List<?> list = handler.executeUpdateQuery(requestMessage, keysGenerated);
if (poller != null) {
SqlParameterSource sqlQueryParameterSource = sqlParameterSourceFactory
.createParameterSource(requestMessage);
if (keysGenerated) {
if (!list.isEmpty()) {
if (list.size() == 1) {
sqlQueryParameterSource = sqlParameterSourceFactory.createParameterSource(list.get(0));
}
else {
sqlQueryParameterSource = sqlParameterSourceFactory.createParameterSource(list);
}
}
}
list = poller.doPoll(sqlQueryParameterSource);
if (list.isEmpty()) {
return null;
}
}
Object payload = list;
if (list.isEmpty()) {
return null;
}
if (list.size() == 1) {
payload = list.get(0);
}
return MessageBuilder.withPayload(payload).copyHeaders(requestMessage.getHeaders()).build();
}
/**
* Flag to indicate that the update query is an insert with autogenerated keys, which will be logged at debug level.
* @param keysGenerated the flag value to set
*/
public void setKeysGenerated(boolean keysGenerated) {
this.keysGenerated = keysGenerated;
}
public void setRequestSqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) {
handler.setSqlParameterSourceFactory(sqlParameterSourceFactory);
}
public void setReplySqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) {
this.sqlParameterSourceFactory = sqlParameterSourceFactory;
}
public void setRowMapper(RowMapper<?> rowMapper) {
poller.setRowMapper(rowMapper);
}
}

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

@@ -61,7 +61,7 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new ExpressionEvaluatingSqlParameterSourceFactory();
private int maxRowsPerPoll = 0;
private volatile int maxRowsPerPoll = 0;
/**
* Constructor taking {@link DataSource} from which the DB Connection can be
@@ -99,7 +99,7 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
this.updatePerRow = updatePerRow;
}
public void setSqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) {
public void setUpdateSqlParameterSourceFactory(SqlParameterSourceFactory sqlParameterSourceFactory) {
this.sqlParameterSourceFactory = sqlParameterSourceFactory;
}
@@ -108,7 +108,7 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
*
* @param sqlQueryParameterSource the sql query parameter source to set
*/
public void setSqlQueryParameterSource(SqlParameterSource sqlQueryParameterSource) {
public void setSelectSqlParameterSource(SqlParameterSource sqlQueryParameterSource) {
this.sqlQueryParameterSource = sqlQueryParameterSource;
}
@@ -143,7 +143,7 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
* mapped results are returned.
*/
private Object poll() {
List<?> payload = doPoll();
List<?> payload = doPoll(this.sqlQueryParameterSource);
if (payload.size() < 1) {
payload = null;
}
@@ -165,7 +165,7 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
this.jdbcOperations.update(this.updateSql, updateParamaterSource);
}
private List<?> doPoll() {
protected List<?> doPoll(SqlParameterSource sqlQueryParameterSource) {
List<?> payload = null;
final RowMapper<?> rowMapper = this.rowMapper == null ? new ColumnMapRowMapper() : this.rowMapper;
@@ -190,9 +190,9 @@ public class JdbcPollingChannelAdapter implements MessageSource<Object> {
resultSetExtractor = temp;
}
if (this.sqlQueryParameterSource != null) {
if (sqlQueryParameterSource != null) {
payload = this.jdbcOperations.getNamedParameterJdbcOperations().query(this.selectQuery,
this.sqlQueryParameterSource, resultSetExtractor);
sqlQueryParameterSource, resultSetExtractor);
}
else {
payload = this.jdbcOperations.getJdbcOperations().query(this.selectQuery, resultSetExtractor);

4
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java
View File

@@ -54,10 +54,10 @@ public class JdbcMessageHandlerParser extends AbstractOutboundChannelAdapterPars
}
String query = IntegrationNamespaceUtils.getTextFromAttributeOrNestedElement(element, "query", parserContext);
if (!StringUtils.hasText(query)) {
throw new BeanCreationException("The query attrbitue is required");
throw new BeanCreationException("The query attribute is required");
}
if (!StringUtils.hasText(query)) {
throw new BeanCreationException("The query attrbitue is required");
throw new BeanCreationException("The query attribute is required");
}
if (refToDataSourceSet) {
builder.addConstructorArgReference(dataSourceRef);

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

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

93
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParser.java Normal file
View File

@@ -0,0 +1,93 @@
/*
* 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.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Dave Syer
* @since 2.0
*
*/
public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
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-gateway", element);
}
String selectQuery = IntegrationNamespaceUtils.getTextFromAttributeOrNestedElement(element, "query",
parserContext);
if (!StringUtils.hasText(selectQuery)) {
selectQuery = null;
}
String updateQuery = IntegrationNamespaceUtils.getTextFromAttributeOrNestedElement(element, "update",
parserContext);
if (!StringUtils.hasText(updateQuery)) {
parserContext.getReaderContext().error("The update attribute is required", element);
return null;
}
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition("org.springframework.integration.jdbc.JdbcOutboundGateway");
if (refToDataSourceSet) {
builder.addConstructorArgReference(dataSourceRef);
}
else {
builder.addConstructorArgReference(jdbcOperationsRef);
}
builder.getRawBeanDefinition().getConstructorArgumentValues().addIndexedArgumentValue(1, updateQuery);
builder.getRawBeanDefinition().getConstructorArgumentValues().addIndexedArgumentValue(2, selectQuery);
IntegrationNamespaceUtils
.setReferenceIfAttributeDefined(builder, element, "reply-sql-parameter-source-factory");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
"request-sql-parameter-source-factory");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-messages-per-poll");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "keys-generated");
String replyChannel = element.getAttribute("reply-channel");
if (StringUtils.hasText(replyChannel)) {
builder.addPropertyReference("outputChannel", replyChannel);
}
return builder;
}
@Override
protected String getInputChannelAttributeName() {
return "request-channel";
}
}

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

@@ -68,8 +68,8 @@ public class JdbcPollingChannelAdapterParser extends AbstractPollingInboundChann
}
builder.addConstructorArgValue(query);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "row-mapper");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-parameter-source-factory");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "sql-query-parameter-source");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "update-sql-parameter-source-factory");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "select-sql-parameter-source");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-rows-per-poll");
if (update!=null) {
builder.addPropertyValue("updateSql", update);

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

@@ -174,13 +174,10 @@
<xsd:documentation>
Reference to a SqlParameterSourceFactory. The input is the result of the
query. The
default factory creates a bean
property parameter source that treats a List in a special
way: the List is
assumed to contain entities with a field called
"id" and these are collected and copied to a field in the
parameter
source called "idList".
default factory creates a parameter source that treats a List in a special
way: the parameter name is used as an expression and projected onto the list,
so for instance "update foos set status=1 where id in (:id)" will generate
an in clause from the properties "id" of the input list elements.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.jdbc.SqlParameterSourceFactory" />
@@ -188,7 +185,7 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="select-sql-query-parameter-source" type="xsd:string">
<xsd:attribute name="select-sql-parameter-source" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
@@ -249,6 +246,15 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="keys-generated" type="xsd:boolean">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Flag to say whether primary keys are generated by the query.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -260,7 +266,13 @@
Defines an outbound Channel Gateway for updating a
database in response to a message on the request
channel and getting a response
on the reply channel.
on the reply channel. The response can be created from a query
supplied here, or (if keys-generated="true") can be the
primary keys generated from an auto-increment, or else just a
count of the number of rows affected by the update. The response
is in general a case insensitive Map (or list of maps if multi-valued), unless
a select query and a row-mapper is provided. If the update count is
returned then the map key is "UPDATE".
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -343,6 +355,18 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-timeout" type="xsd:string" />
<xsd:attribute name="keys-generated" type="xsd:boolean">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Flag to say whether primary keys are generated by the query. If they are then
they can be used as a reply payload instead of providing select query. A single
valued result is extracted before returning (the usual case), so the payload of the reply message
can be a Map (column name to value) or a list of maps.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

2
spring-integration-jdbc/src/test/java/log4j.properties
View File

@@ -7,5 +7,5 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m
log4j.category.org.springframework=WARN
log4j.category.org.springframework.integration=DEBUG
log4j.category.org.springframework.integration.jdbc=WARN
log4j.category.org.springframework.integration.jdbc=DEBUG
log4j.category.org.springframework.jdbc=DEBUG

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java
View File

@@ -88,7 +88,7 @@ public class JdbcPollingChannelAdapterIntegrationTests {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
this.embeddedDatabase,
"select * from item where status=:status");
adapter.setSqlQueryParameterSource(new SqlParameterSource() {
adapter.setSelectSqlParameterSource(new SqlParameterSource() {
public boolean hasValue(String name) {
return "status".equals(name);

95
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java Normal file
View File

@@ -0,0 +1,95 @@
package org.springframework.integration.jdbc.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
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.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
public class JdbcOutboundGatewayParserTests {
private SimpleJdbcTemplate jdbcTemplate;
private MessageChannel channel;
private ConfigurableApplicationContext context;
private MessagingTemplate messagingTemplate;
@Test
public void testMapPayloadMapReply() {
setUp("handlingMapPayloadJdbcOutboundGatewayTest.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 name", "bar", map.get("name"));
Message<?> reply = messagingTemplate.receive();
assertNotNull(reply);
@SuppressWarnings("unchecked")
Map<String, ?> payload = (Map<String, ?>) reply.getPayload();
assertEquals("bar", payload.get("name"));
}
@Test
public void testKeyGeneration() {
setUp("handlingKeyGenerationJdbcOutboundGatewayTest.xml", getClass());
Message<?> message = MessageBuilder.withPayload(Collections.singletonMap("foo", "bar")).build();
channel.send(message);
Message<?> reply = messagingTemplate.receive();
assertNotNull(reply);
@SuppressWarnings("unchecked")
Map<String, ?> payload = (Map<String, ?>) reply.getPayload();
Object id = payload.get("SCOPE_IDENTITY()");
assertNotNull(id);
Map<String, Object> map = this.jdbcTemplate.queryForMap("SELECT * from BARS");
assertEquals("Wrong id", id, map.get("ID"));
assertEquals("Wrong name", "bar", map.get("name"));
}
@Test
public void testCountUpdates() {
setUp("handlingCountUpdatesJdbcOutboundGatewayTest.xml", getClass());
Message<?> message = MessageBuilder.withPayload(Collections.singletonMap("foo", "bar")).build();
channel.send(message);
Message<?> reply = messagingTemplate.receive();
assertNotNull(reply);
@SuppressWarnings("unchecked")
Map<String, ?> payload = (Map<String, ?>) reply.getPayload();
assertEquals(1, payload.get("updated"));
}
@After
public void tearDown() {
if (context != null) {
context.close();
}
}
protected void setupMessagingTemplate() {
PollableChannel pollableChannel = this.context.getBean("output", PollableChannel.class);
this.messagingTemplate = new MessagingTemplate(pollableChannel);
this.messagingTemplate.setReceiveTimeout(500);
}
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);
setupMessagingTemplate();
}
}

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

@@ -0,0 +1,21 @@
<?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">
<si:channel id="output">
<si:queue />
</si:channel>
<outbound-gateway update="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])" request-channel="target"
reply-channel="output" data-source="dataSource" />
<beans:import resource="jdbcOutboundChannelAdapterCommonConfig.xml" />
</beans:beans>

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

@@ -0,0 +1,21 @@
<?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">
<si:channel id="output">
<si:queue />
</si:channel>
<outbound-gateway update="insert into bars (status, name) values (0, :payload[foo])" request-channel="target"
reply-channel="output" data-source="dataSource" keys-generated="true" />
<beans:import resource="jdbcOutboundChannelAdapterCommonConfig.xml" />
</beans:beans>

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

@@ -0,0 +1,21 @@
<?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">
<si:channel id="output">
<si:queue />
</si:channel>
<outbound-gateway query="select * from foos where id=:headers[$id]" update="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])"
request-channel="target" reply-channel="output" data-source="dataSource" />
<beans:import resource="jdbcOutboundChannelAdapterCommonConfig.xml" />
</beans:beans>

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcOutboundChannelAdapterCommonConfig.xml
View File

@@ -9,7 +9,7 @@
<si:channel id="target"/>
<jdbc:embedded-database type="HSQL" id="dataSource">
<jdbc:embedded-database type="H2" id="dataSource">
<jdbc:script location="org/springframework/integration/jdbc/config/outboundSchema.sql" />
</jdbc:embedded-database>

3
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundSchema.sql
View File

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

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml
View File

@@ -11,7 +11,7 @@
<inbound-channel-adapter query="select * from item where status=2" channel="target"
update="update item set status=1, name=:foo where id in (:id)" jdbc-operations="jdbcTemplate"
sql-parameter-source-factory="sqlParameterSourceFactory" />
update-sql-parameter-source-factory="sqlParameterSourceFactory" />
<beans:import resource="jdbcInboundChannelAdapterCommonConfig.xml" />

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml
View File

@@ -11,7 +11,7 @@
http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd">
<inbound-channel-adapter query="select * from item where status=:status" channel="target"
data-source="dataSource" sql-query-parameter-source="parameterSource"/>
data-source="dataSource" select-sql-parameter-source="parameterSource"/>
<beans:import resource="jdbcInboundChannelAdapterCommonConfig.xml" />