diff --git a/spring-integration-jdbc/pom.xml b/spring-integration-jdbc/pom.xml index b8a8eadfc1..64de7a0f75 100644 --- a/spring-integration-jdbc/pom.xml +++ b/spring-integration-jdbc/pom.xml @@ -79,6 +79,12 @@ 10.5.3.0_1 test + + com.h2database + h2 + 1.2.125 + test + @@ -97,8 +103,7 @@ - + @@ -111,8 +116,7 @@ - + @@ -125,8 +129,7 @@ - + @@ -139,8 +142,7 @@ - + @@ -153,8 +155,7 @@ - + @@ -167,8 +168,7 @@ - + @@ -181,8 +181,7 @@ - + @@ -195,8 +194,7 @@ - + @@ -209,8 +207,7 @@ - + @@ -262,7 +259,9 @@ repository.objectstyle ObjectStyle.org Repository http://objectstyle.org/maven2/ - false + + false + diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java index 3ad92ddd5e..8a76174132 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java @@ -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); } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java index 9176a9a265..0e56d96c28 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java @@ -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> keys = executeUpdateQuery(message, keysGenerated); + if (logger.isDebugEnabled() && !keys.isEmpty()) { + logger.debug("Generated keys: "+keys); } } + protected List> 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 map = new LinkedCaseInsensitiveMap(); + map.put("UPDATED", updated); + return Collections.singletonList(map); + } + + } + } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java new file mode 100644 index 0000000000..252985cbba --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java @@ -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); + } + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java index 7703d637b6..5a9b4967f1 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java @@ -61,7 +61,7 @@ public class JdbcPollingChannelAdapter implements MessageSource { 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 { 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 { * * @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 { * 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 { 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 { 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); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java index ecd81ba4db..88f6a3cb66 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParser.java @@ -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); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcNamespaceHandler.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcNamespaceHandler.java index b9ca9a6b7d..3b00d75b94 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcNamespaceHandler.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcNamespaceHandler.java @@ -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()); } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParser.java new file mode 100644 index 0000000000..d3fa1c2e46 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParser.java @@ -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"; + } +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java index e324353693..d774d87381 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParser.java @@ -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); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd index 6179338d01..4ce14d0b29 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd @@ -174,13 +174,10 @@ 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. @@ -188,7 +185,7 @@ - + @@ -249,6 +246,15 @@ + + + + + Flag to say whether primary keys are generated by the query. + + + + @@ -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". @@ -343,6 +355,18 @@ + + + + + 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. + + + + diff --git a/spring-integration-jdbc/src/test/java/log4j.properties b/spring-integration-jdbc/src/test/java/log4j.properties index a855018c87..1dcc129804 100644 --- a/spring-integration-jdbc/src/test/java/log4j.properties +++ b/spring-integration-jdbc/src/test/java/log4j.properties @@ -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 diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java index 3ce97ce50c..4b64698941 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java @@ -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); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java new file mode 100644 index 0000000000..5bbeec983c --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java @@ -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 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 payload = (Map) 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 payload = (Map) reply.getPayload(); + Object id = payload.get("SCOPE_IDENTITY()"); + assertNotNull(id); + Map 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 payload = (Map) 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(); + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingCountUpdatesJdbcOutboundGatewayTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingCountUpdatesJdbcOutboundGatewayTest.xml new file mode 100644 index 0000000000..6dea01de64 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingCountUpdatesJdbcOutboundGatewayTest.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingKeyGenerationJdbcOutboundGatewayTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingKeyGenerationJdbcOutboundGatewayTest.xml new file mode 100644 index 0000000000..f6ded8d4a6 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingKeyGenerationJdbcOutboundGatewayTest.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingMapPayloadJdbcOutboundGatewayTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingMapPayloadJdbcOutboundGatewayTest.xml new file mode 100644 index 0000000000..bbe17af91e --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/handlingMapPayloadJdbcOutboundGatewayTest.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcOutboundChannelAdapterCommonConfig.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcOutboundChannelAdapterCommonConfig.xml index 1b5b5f52e4..39399399ab 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcOutboundChannelAdapterCommonConfig.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcOutboundChannelAdapterCommonConfig.xml @@ -9,7 +9,7 @@ - + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundSchema.sql b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundSchema.sql index 638755ab0c..ab874aa183 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundSchema.sql +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundSchema.sql @@ -1 +1,2 @@ -create table foos(id varchar(100),status int,name varchar(20)); \ No newline at end of file +create table foos(id varchar(100),status int,name varchar(20)); +create table bars(id int identity,status int,name varchar(20)); \ No newline at end of file diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml index 721d98bcc6..2a6afc85b3 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParameterSourceJdbcInboundChannelAdapterTest.xml @@ -11,7 +11,7 @@ + update-sql-parameter-source-factory="sqlParameterSourceFactory" /> diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml index b80a751dd5..30d76d9295 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/pollingWithParametersForMapJdbcInboundChannelAdapterTest.xml @@ -11,7 +11,7 @@ http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd"> + data-source="dataSource" select-sql-parameter-source="parameterSource"/> diff --git a/src/docbkx/jdbc.xml b/src/docbkx/jdbc.xml index ac4ab9cfd0..9ee85725a1 100644 --- a/src/docbkx/jdbc.xml +++ b/src/docbkx/jdbc.xml @@ -36,16 +36,15 @@ the next poll. The update can be parameterised by the list of ids from the original select. This is done through a naming convention by default (a column in the input result set called "id" is translated into a list in - the parameter map for the update called "id"). The following example defines - an inbound Channel Adapter with an update query and a DataSource - reference. DataSource reference. <jdbc:inbound-channel-adapter query="select * from item where status=2" channel="target" data-source="dataSource" - update="update item set status=10 where id in (:id)" />]]> + update="update item set status=10 where id in (:id)" /> - The parameters in the update query are specified with a colon (:) prefix to the name of a parameter (which in this case is an expression to be applied to each of the rows in the polled result set). This is a standard feature of the named parameter JDBC support in Spring JDBC combined with a convention (projection onto the polled result list) adopted in Spring Integration. The underlying Spring JDBC features limit the available expressions (e.g. most special characters other than period are disallowed), but since the target is usually a list of or an individual object addressable by simple bean paths this isn't unduly restrictive. - - To change the parameter - generation strategy you can inject a + The parameters in the update query are specified with a colon (:) prefix to the name of a parameter (which in this case is an expression to be applied to each of the rows in the polled result set). This is a standard feature of the named parameter JDBC support in Spring JDBC combined with a convention (projection onto the polled result list) adopted in Spring Integration. The underlying Spring JDBC features limit the available expressions (e.g. most special characters other than period are disallowed), but since the target is usually a list of or an individual object addressable by simple bean paths this isn't unduly restrictive. + To change the parameter generation strategy you can inject a SqlParameterSourceFactory into the adapter to override the default behaviour (the adapter has a sql-parameter-source-factory attribute). @@ -58,14 +57,14 @@ controlled. A very important feature of the poller for JDBC usage is the option to wrap the poll operation in a transaction, for example: - <jdbc:inbound-channel-adapter query="..." channel="target" data-source="dataSource" - update="..."> - - - - -]]> + update="..."> + <poller> + <interval-trigger interval="1000"/> + <transactional/> + </poller> +</jdbc:inbound-channel-adapter> If a poller is not explicitly specified a default value will be used (and as per normal with Spring Integration can be defined as a top level bean) @@ -87,17 +86,21 @@ The outbound Channel Adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query. The message payload and headers are available by default as input parameters to the - query, for instance: <jdbc:outbound-channel-adapter query="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])" - channel="input" data-source="dataSource"/>]]> In the + channel="input" data-source="dataSource"/> In the example above, messages arriving on the channel "input" have a payload of a map with key "foo", so the [] operator dereferences that value from the map. The headers are also accessed as a map. The parameters in the query above are bean property expressions on the incoming message (not Spring EL expressions). This behaviour is part of the - SqlParameterSource which is the default - source created by the outbound adapter. Other behaviour is possible - in the adapter, and requires the user to inject a different - SqlParameterSourceFactory. + + SqlParameterSource + + which is the default source created by the outbound adapter. Other behaviour is possible in the adapter, and requires the user to inject a different + + SqlParameterSourceFactory + + . The outbound adapter requires a reference to either a DataSource or @@ -110,6 +113,55 @@ there is one) as the sender of the message. +
+ Outbound Gateway + + The outbound Gateway is like a combination of the outbound and + inbound adapters: its role is to handle a message and use it to execute a + SQL query and then respond with the result sending it to a reply channel. + The message payload and headers are available by default as input + parameters to the query, for instance: <jdbc:outbound-gateway + update="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])" + request-channel="input" reply-channel="output" data-source="dataSource" /> + + The result of the above would be to insert a record into the "foos" + table and return a message to the output channel indicating the number of + rows affected (the payload is a map {UPDATED=1}. + + If the update query is an insert with auto-generated keys, the reply + message can be populated with the generated keys by adding + keys-generated="true" to the above example (this is not + the default because it is not supported by some database platforms). For + example: + + <jdbc:outbound-gateway + update="insert into foos (status, name) values (0, :payload[foo])" + request-channel="input" reply-channel="output" data-source="dataSource" + keys-generated="true"/> + + Instead of the update count or the generated keys, you can also + provide a select query to execute and generate a reply message that way + (like the inbound adapter), e.g: + + <jdbc:outbound-gateway + update="insert into foos (id, status, name) values (:headers[$id], 0, :payload[foo])" + query="select * from foos where id=:headers[$id]" + request-channel="input" reply-channel="output" data-source="dataSource" /> + + Like with the adapters there is also the option to provide + SqlParameterSourceFactory instances for request and + reply. The default is the same as for the outbound adapter, so the request + message is available as the root of an expression. If + keys-generated="true" then the root of the expression is the generated + keys (a map if there is only one or a list of maps if + multi-valued). + + The outbound gateway requires a reference to either a DataSource or + a JdbcTemplate. It can also have a + SqlParameterSourceFactory injected to control the + binding of incoming message to the query. +
+
Message Store @@ -120,15 +172,15 @@ implemented by the JdbcMessageStore and there is also support for configuring store instances in XML. For example: - ]]> + <jdbc:message-store id="messageStore" data-source="dataSource"/> A JdbcTemplate can be specified instead of a DataSource. Other optional attributes are show in the next example: - ]]>Here we + <jdbc:message-store id="messageStore" data-source="dataSource" + lob-handler="lobHandler" table-prefix="MY_INT_"/>Here we have specified a LobHandler for dealing with messages as large objects (e.g. often necessary if using Oracle) and a prefix for the table names in the queries generated by the store. The diff --git a/src/docbkx/jdbc.xml~ b/src/docbkx/jdbc.xml~ new file mode 100644 index 0000000000..fd9e353290 --- /dev/null +++ b/src/docbkx/jdbc.xml~ @@ -0,0 +1,203 @@ + + + + JDBC Support + + Spring Integration provides Channel Adapters for receiving and sending + messages via database queries. + +
+ Inbound Channel Adapter + + The main function of an inbound Channel Adapter is to execute a SQL + SELECT query and turn the result set into a message. The + message payload is the whole result set, expressed as a + List, and the types of the items in the list + depends on the row-mapping strategy that is used. The default strategy is + a generic mapper that just returns a Map for each + row i nthe query. Optionally this can be changed by adding a reference to + requires a reference to a RowMapper instance (see + the Spring + JDBC documentation for more detailed information about row + mapping). + If you want to convert rows in the SELECT query result to + individual messages you can use a downstream splitter. + + + The inbound adapter also requires a reference to either + JdbcTemplate instance or + DataSource. + + As well as the SELECT statement to generate the + messages, the adapter above also has an UPDATE statement that + is being used to mark the records as processed, so they don't show up in + the next poll. The update can be parameterised by the list of ids from the + original select. This is done through a naming convention by default (a + column in the input result set called "id" is translated into a list in + the parameter map for the update called "id"). The following example defines + an inbound Channel Adapter with an update query and a DataSource + reference. ]]> + + The parameters in the update query are specified with a colon (:) prefix to the name of a parameter (which in this case is an expression to be applied to each of the rows in the polled result set). This is a standard feature of the named parameter JDBC support in Spring JDBC combined with a convention (projection onto the polled result list) adopted in Spring Integration. The underlying Spring JDBC features limit the available expressions (e.g. most special characters other than period are disallowed), but since the target is usually a list of or an individual object addressable by simple bean paths this isn't unduly restrictive. + + To change the parameter + generation strategy you can inject a + SqlParameterSourceFactory into the adapter to + override the default behaviour (the adapter has a + sql-parameter-source-factory attribute). + +
+ Polling and Transactions + + The inbound adapter accepts a regular Spring Integration poller as + a sub element, so for instance the frequency of the polling can be + controlled. A very important feature of the poller for JDBC usage is the + option to wrap the poll operation in a transaction, for example: + + + + + + +]]> + + + If a poller is not explicitly specified a default value will be used (and as per normal with Spring Integration can be defined as a top level bean) + In this example the database is polled every 1000 + milliseconds, and the update and select queries are both executed in the + same transaction. The transaction manager configuration is not shown, + but as long as it is aware of the data source then the poll is + transactional. A common use case is for the downstream channels to be + direct channels (the default), so that the endpoints are invoked in the + same thread, and hence the same transaction. then if any of them fails, + the transaction rolls back and the input data are reverted to their + original state. +
+
+ +
+ Outbound Channel Adapter + + The outbound Channel Adapter is the inverse of the inbound: its role + is to handle a message and use it to execute a SQL query. The message + payload and headers are available by default as input parameters to the + query, for instance: ]]> In the + example above, messages arriving on the channel "input" have a payload of + a map with key "foo", so the [] operator dereferences that + value from the map. The headers are also accessed as a map. + The parameters in the query above are bean property expressions on the incoming message (not Spring EL expressions). This behaviour is part of the + SqlParameterSource which is the default + source created by the outbound adapter. Other behaviour is possible + in the adapter, and requires the user to inject a different + SqlParameterSourceFactory. + + + The outbound adapter requires a reference to either a DataSource or + a JdbcTemplate. It can also have a + SqlParameterSourceFactory injected to control the + binding of incoming message to the query. + + If the input channel is a direct channel then the outbound adapter + runs its query in the same thread, and therefor ethe same transaction (if + there is one) as the sender of the message. +
+ +
+ Outbound Gateway + + The outbound Gateway is like a combination of the inbound and outbound adapters: its role + is to handle a message and use it to execute a SQL query. The message + payload and headers are available by default as input parameters to the + query, for instance: ]]> In the + example above, messages arriving on the channel "input" have a payload of + a map with key "foo", so the [] operator dereferences that + value from the map. The headers are also accessed as a map. + The parameters in the query above are bean property expressions on the incoming message (not Spring EL expressions). This behaviour is part of the + SqlParameterSource which is the default + source created by the outbound adapter. Other behaviour is possible + in the adapter, and requires the user to inject a different + SqlParameterSourceFactory. + + + The outbound adapter requires a reference to either a DataSource or + a JdbcTemplate. It can also have a + SqlParameterSourceFactory injected to control the + binding of incoming message to the query. + + If the input channel is a direct channel then the outbound adapter + runs its query in the same thread, and therefor ethe same transaction (if + there is one) as the sender of the message. +
+ +
+ Message Store + + The JDBC module provides an implementation of the Spring Integration + MessageStore (important in the Claim Check pattern) + and MessageGroupStore (important in stateful + patterns like Aggregator) backed by a database. Both interfaces are + implemented by the JdbcMessageStore and there is also support for + configuring store instances in XML. For example: + + ]]> + + A JdbcTemplate can be specified instead of a + DataSource. + + Other optional attributes are show in the next example: + + ]]>Here we + have specified a LobHandler for dealing with + messages as large objects (e.g. often necessary if using Oracle) and a + prefix for the table names in the queries generated by the store. The + table name prefix defaults to "INT_". + +
+ Initializing the Database + + Spring Integration ships with some sample scripts that can be used + to initialize a database. In the spring-integration-jdbc JAR file you + will find scripts in the + org.springframework.integration.jdbc package: + there is a create and a drop script example for a range of common + database platforms. A common way to use these scripts is to reference + them in a Spring + JDBC data source initializer. Note that the scripts are provided + as samples or specifications of the the required table and column names. + You may find that you need to enhance them for production use (e.g. with + index declarations). +
+ +
+ Partitioning a Message Store + + It is common to use a JdbcMessageStore as a + global store for a group of applications, or nodes in the same + application. To provide some portection against name clashes, and to + give control over the database meta-data configuration, the message + store allows the tables to be partitioned in two ways. One is to use + separate table names, by changing the prefix as described above, and the + other is to specify a "region" name for partitioning data within a + single table. An important use case for this is using the store to + manage persistent queues backing a Spring Integration channel. The + message data for a persistent channel is keyed in the store on the + channel name, so if the channel names are not globally unique then there + is the danger of channels picking up data that was not intended for + them. To avoid this the message store region can be used to keep data + separate for different physical channels that happen to have the same + logical name. +
+
+