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

INT-2246 - Add attribute "max-rows-per-poll" to Jdbc Outbound Gateway

Browse Source 
For reference see: https://jira.springsource.org/browse/INT-2246
This commit is contained in:
Gunnar Hillert
2011-11-16 15:43:15 -05:00
committed by Mark Fisher
parent 910675be39
commit 6f33252c7c
8 changed files with 174 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -27,10 +27,11 @@ import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*
* @since 2.0
*/
public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler implements InitializingBean {
@@ -67,7 +68,10 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
}
public void setMaxRowsPerPoll(int maxRows) {
Assert.notNull(poller, "If you want to set 'maxRowsPerPoll', then you must provide a 'selectQuery'.");
poller.setMaxRowsPerPoll(maxRows);
}
@Override

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

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 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.
@@ -24,7 +24,7 @@ import org.w3c.dom.Element;
/**
* @author Dave Syer
* @since 2.0
*
*
*/
public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
@@ -75,7 +75,7 @@ public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
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, "max-rows-per-poll");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "keys-generated");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");

24
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.1.xsd
View File

@@ -171,13 +171,11 @@
</xsd:attribute>
<xsd:attribute name="max-rows-per-poll" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Limits the number of rows extracted per query (otherwise all rows
are extracted into the
outgoing message).
</xsd:documentation>
</xsd:appinfo>
<xsd:documentation>
Limits the number of rows extracted per query (otherwise all rows
are extracted into the
outgoing message).
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string" use="required">
@@ -361,6 +359,18 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="max-rows-per-poll" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
When using a select query, you can set a
custom limit regarding the number of rows
extracted. Otherwise by default only the first
row will be extracted into the outgoing message.
If set to '0' all rows are extracted.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-sql-parameter-source-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

55
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcOutboundGatewayTests.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2002-2011 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 static org.junit.Assert.fail;
import javax.sql.DataSource;
import static junit.framework.Assert.assertEquals;
import org.junit.Test;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
/**
*
* @author Gunnar Hillert
* @since 2.1
*
*/
public class JdbcOutboundGatewayTests {
/**
* Test method for
* {@link org.springframework.integration.jdbc.JdbcOutboundGateway#setMaxRowsPerPoll(int)}.
*/
@Test
public void testSetMaxRowsPerPollWithoutSelectQuery() {
DataSource dataSource = new EmbeddedDatabaseBuilder().build();
JdbcOutboundGateway jdbcOutboundGateway = new JdbcOutboundGateway(dataSource, "select * from DOES_NOT_EXIST");
try {
jdbcOutboundGateway.setMaxRowsPerPoll(10);
} catch (IllegalArgumentException e) {
assertEquals("If you want to set 'maxRowsPerPoll', then you must provide a 'selectQuery'.", e.getMessage());
return;
}
fail("Expected an IllegalArgumentException to be thrown.");
}
}

10
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundAdapterWithPollerTest-context.xml
View File

@@ -9,11 +9,11 @@
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xmlns:jdbc="http://www.springframework.org/schema/jdbc">
<int:channel id="target">
<int:queue/>
</int:channel>
<int:channel id="output">
<int:queue />
</int:channel>
@@ -24,15 +24,15 @@
</int-jdbc:outbound-channel-adapter>
<jdbc:embedded-database type="H2" id="dataSource"/>
<jdbc:initialize-database ignore-failures="DROPS">
<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
<jdbc:script location="org/springframework/integration/jdbc/config/outboundPollerSchema.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>

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

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 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.
@@ -31,7 +31,6 @@ 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.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.jdbc.JdbcOutboundGateway;
import org.springframework.integration.support.MessageBuilder;
@@ -60,7 +59,7 @@ public class JdbcOutboundGatewayParserTests {
setUp("handlingMapPayloadJdbcOutboundGatewayTest.xml", getClass());
assertTrue(context.containsBean("jdbcGateway"));
Message<?> message = MessageBuilder.withPayload(Collections.singletonMap("foo", "bar")).build();
channel.send(message);
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"));
@@ -139,6 +138,40 @@ public class JdbcOutboundGatewayParserTests {
}
@Test
public void testDefaultMaxMessagesPerPollIsSet() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("JdbcOutboundGatewayWithPollerTest-context.xml", this.getClass());
PollingConsumer pollingConsumer = ac.getBean(PollingConsumer.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(pollingConsumer);
Object source = accessor.getPropertyValue("handler");
accessor = new DirectFieldAccessor(source);
source = accessor.getPropertyValue("poller"); //JdbcPollingChannelAdapter
accessor = new DirectFieldAccessor(source);
Integer maxRowsPerPoll = (Integer) accessor.getPropertyValue("maxRowsPerPoll");
assertEquals("maxRowsPerPoll should default to 1", Integer.valueOf(1), maxRowsPerPoll);
}
@Test
public void testMaxMessagesPerPollIsSet() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("JdbcOutboundGatewayWithPoller2Test-context.xml", this.getClass());
PollingConsumer pollingConsumer = ac.getBean(PollingConsumer.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(pollingConsumer);
Object source = accessor.getPropertyValue("handler");
accessor = new DirectFieldAccessor(source);
source = accessor.getPropertyValue("poller"); //JdbcPollingChannelAdapter
accessor = new DirectFieldAccessor(source);
Integer maxRowsPerPoll = (Integer) accessor.getPropertyValue("maxRowsPerPoll");
assertEquals("maxRowsPerPoll should default to 10", Integer.valueOf(10), maxRowsPerPoll);
}
@After
public void tearDown() {
if (context != null) {

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

@@ -0,0 +1,40 @@
<?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="target">
<int:queue/>
</int:channel>
<int:channel id="output">
<int:queue />
</int:channel>
<int-jdbc:outbound-gateway query="select * from bazz where id=:headers[id]" update="insert into bazz (id, status, name) values (:headers[id], 0, :payload[foo])"
request-channel="target" reply-channel="output" data-source="dataSource" auto-startup="true" max-rows-per-poll="10">
<int:poller fixed-rate="1000"/>
</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/outboundPollerSchema.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>

19
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPollerTest-context.xml
View File

@@ -9,11 +9,11 @@
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xmlns:jdbc="http://www.springframework.org/schema/jdbc">
<int:channel id="target">
<int:queue/>
</int:channel>
<int:channel id="output">
<int:queue />
</int:channel>
@@ -22,18 +22,17 @@
request-channel="target" reply-channel="output" data-source="dataSource" auto-startup="true" reply-timeout="444">
<int:poller fixed-rate="1000"/>
</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/outboundPollerSchema.sql"/>
</jdbc:initialize-database>
<jdbc:embedded-database type="H2" id="dataSource"/>
<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
<jdbc:script location="org/springframework/integration/jdbc/config/outboundPollerSchema.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>