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

INT-2289 - Make update optional for JDBC Outbound Gateway

Browse Source 
For reference: https://jira.springsource.org/browse/INT-2289
This commit is contained in:
Gunnar Hillert
2012-06-27 11:29:51 -04:00
committed by Gary Russell
parent 884a4e5bb3
commit bc244eff8f
9 changed files with 190 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.integration.jdbc;
import java.util.Collections;
import java.util.List;
import javax.sql.DataSource;
@@ -28,6 +29,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Dave Syer
@@ -60,14 +62,28 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
}
public JdbcOutboundGateway(JdbcOperations jdbcOperations, String updateQuery, String selectQuery) {
if (selectQuery != null) {
Assert.notNull(jdbcOperations, "'jdbcOperations' must not be null.");
if (!StringUtils.hasText(updateQuery) && !StringUtils.hasText(selectQuery)) {
throw new IllegalArgumentException("The 'updateQuery' and the 'selectQuery' must not both be null or empty.");
}
if (StringUtils.hasText(selectQuery)) {
poller = new JdbcPollingChannelAdapter(jdbcOperations, selectQuery);
poller.setMaxRowsPerPoll(1);
}
else {
poller = null;
}
handler = new JdbcMessageHandler(jdbcOperations, updateQuery);
if (StringUtils.hasText(updateQuery)) {
handler = new JdbcMessageHandler(jdbcOperations, updateQuery);
}
else {
handler = null;
}
}
/**
@@ -96,13 +112,24 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
poller.setMaxRowsPerPoll(this.maxRowsPerPoll);
}
handler.afterPropertiesSet();
if (this.handler!= null) {
handler.afterPropertiesSet();
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
List<?> list = handler.executeUpdateQuery(requestMessage, keysGenerated);
List<?> list;
if (this.handler != null) {
list = handler.executeUpdateQuery(requestMessage, keysGenerated);
}
else {
list = Collections.emptyList();
}
if (poller != null) {
SqlParameterSource sqlQueryParameterSource = sqlParameterSourceFactory
.createParameterSource(requestMessage);

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

@@ -36,22 +36,18 @@ public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
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(JdbcOutboundGateway.class);
if (refToDataSourceSet) {
@@ -61,8 +57,8 @@ public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
builder.addConstructorArgReference(jdbcOperationsRef);
}
builder.getRawBeanDefinition().getConstructorArgumentValues().addIndexedArgumentValue(1, updateQuery);
builder.getRawBeanDefinition().getConstructorArgumentValues().addIndexedArgumentValue(2, selectQuery);
builder.addConstructorArgValue(updateQuery);
builder.addConstructorArgValue(selectQuery);
IntegrationNamespaceUtils
.setReferenceIfAttributeDefined(builder, element, "reply-sql-parameter-source-factory");

30
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.2.xsd
View File

@@ -315,9 +315,14 @@
<xsd:annotation>
<xsd:documentation>
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. The response can be created from a query
database in response to a message on the request channel, and/or
for retrieving data from the database using the input message as
a source of parameters for the specified SQL select query.
The database response will be used to create the response Message
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
@@ -334,9 +339,19 @@
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
An update query to execute when a message is
An update query to be executed when a message is
received. If this is in a transaction then the
update will roll back when the transaction does.
The update can also be specified using the
"update" attribute.
Since Spring Integration 2.2 specifying an
update query is optional, if at least the
select query is specified.
If you specify both, update- and select
query, then the update is executed first.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
@@ -353,6 +368,13 @@
update will roll back when the transaction does.
The update can also be specified as a nested element.
Since Spring Integration 2.2 specifying an
update query is optional, if at least the
select query is specified.
If you specify both, update- and select
query, then the update is executed first.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>

42
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcOutboundGatewayTests.java
View File

@@ -16,9 +16,12 @@ import static org.junit.Assert.fail;
import javax.sql.DataSource;
import junit.framework.Assert;
import static junit.framework.Assert.assertEquals;
import org.junit.Test;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
/**
@@ -35,12 +38,12 @@ public class JdbcOutboundGatewayTests {
DataSource dataSource = new EmbeddedDatabaseBuilder().build();
JdbcOutboundGateway jdbcOutboundGateway = new JdbcOutboundGateway(dataSource, "select * from DOES_NOT_EXIST");
JdbcOutboundGateway jdbcOutboundGateway = new JdbcOutboundGateway(dataSource, "update something");
try {
jdbcOutboundGateway.setMaxRowsPerPoll(10);
jdbcOutboundGateway.onInit();
} catch (IllegalArgumentException e) {
assertEquals("If you want to set 'maxRowsPerPoll', then you must provide a 'selectQuery'.", e.getMessage());
return;
@@ -50,6 +53,41 @@ public class JdbcOutboundGatewayTests {
}
@Test
public void testConstructorWithNulljdbcOperations() {
JdbcOperations jdbcOperations = null;
try {
new JdbcOutboundGateway(jdbcOperations, "select * from DOES_NOT_EXIST");
}
catch (IllegalArgumentException e) {
Assert.assertEquals("'jdbcOperations' must not be null.", e.getMessage());
return;
}
fail("Expected an IllegalArgumentException to be thrown.");
}
@Test
public void testConstructorWithEmptyAndNullQueries() {
final DataSource dataSource = new EmbeddedDatabaseBuilder().build();
final String selectQuery = " ";
final String updateQuery = null;
try {
new JdbcOutboundGateway(dataSource, updateQuery, selectQuery);
}
catch (IllegalArgumentException e) {
Assert.assertEquals("The 'updateQuery' and the 'selectQuery' must not both be null or empty.", e.getMessage());
return;
}
fail("Expected an IllegalArgumentException to be thrown.");
}
/**
* Test method for
* {@link org.springframework.integration.jdbc.JdbcOutboundGateway#setMaxRowsPerPoll(Integer)}.

22
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java
View File

@@ -119,6 +119,28 @@ public class JdbcOutboundGatewayParserTests {
assertEquals("bar", payload.get("name"));
}
@Test
public void testWithSelectQueryOnly() throws Exception{
ApplicationContext ac = new ClassPathXmlApplicationContext("JdbcOutboundGatewayWithSelectTest-context.xml", this.getClass());
Message<?> message = MessageBuilder.withPayload(Integer.valueOf(100)).build();
MessageChannel requestChannel = ac.getBean("request", MessageChannel.class);
PollableChannel replyChannel = ac.getBean("reply", PollableChannel.class);
requestChannel.send(message);
Thread.sleep(1000);
@SuppressWarnings("unchecked")
Message<Map<String, Object>> reply = (Message<Map<String, Object>>) replyChannel.receive(500);
String id = (String) reply.getPayload().get("id");
Integer status = (Integer) reply.getPayload().get("status");
String name = (String) reply.getPayload().get("name");
assertEquals("100", id);
assertEquals(Integer.valueOf(3), status);
assertEquals("Cartman", name);
}
@Test
public void testReplyTimeoutIsSet() throws Exception {
setUp("JdbcOutboundGatewayWithPollerTest-context.xml", getClass());

36
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithSelectTest-context.xml Normal file
View File

@@ -0,0 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xmlns:jdbc="http://www.springframework.org/schema/jdbc">
<int:channel id="request"/>
<int:channel id="reply">
<int:queue/>
</int:channel>
<int-jdbc:outbound-gateway id="jdbcOutboundGateway" query="select * from bazz where id=:payload"
request-channel="request" reply-channel="reply" data-source="dataSource" auto-startup="true" reply-timeout="444">
</int-jdbc:outbound-gateway>
<jdbc:embedded-database id="dataSource" type="H2"/>
<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
<jdbc:script location="classpath:org/springframework/integration/jdbc/config/outboundPollerSchemaWithData.sql"/>
</jdbc:initialize-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>

5
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/outboundPollerSchemaWithData.sql Normal file
View File

@@ -0,0 +1,5 @@
drop table bazz;
create table bazz(id varchar(100),status int,name varchar(20));
INSERT INTO bazz (id, status, name) VALUES (100, 3, 'Cartman')