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

Merge pull request #187 from ghillert/INT-2246

Browse Source 
Add attribute "max-rows-per-poll" to Jdbc Outbound Gateway

  For reference see: https://jira.springsource.org/browse/INT-2246
This commit is contained in:
Mark Fisher
2011-11-22 15:43:11 -05:00
parent 910675be39 462af6bea0
commit 51e706983f
8 changed files with 267 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -27,10 +27,12 @@ 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
*
* @author Gunnar Hillert
*
* @since 2.0
*/
public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler implements InitializingBean {
@@ -43,6 +45,8 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
private volatile boolean keysGenerated;
private volatile Integer maxRowsPerPoll;
public JdbcOutboundGateway(DataSource dataSource, String updateQuery) {
this(new JdbcTemplate(dataSource), updateQuery, null);
}
@@ -66,13 +70,34 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
handler = new JdbcMessageHandler(jdbcOperations, updateQuery);
}
public void setMaxRowsPerPoll(int maxRows) {
poller.setMaxRowsPerPoll(maxRows);
/**
* The maximum number of rows to pull out of the query results per poll (if
* greater than zero, otherwise all rows will be packed into the outgoing
* message).
*
* The value is ultimately set on the underlying {@link JdbcPollingChannelAdapter}.
* If not specified this value will default to <code>zero</code>.
*
* This parameter is only applicable if a selectQuery was provided. Null values
* are not permitted.
*
* @param maxRowsPerPoll Must not be null.
*/
public void setMaxRowsPerPoll(Integer maxRowsPerPoll) {
Assert.notNull(maxRowsPerPoll, "MaxRowsPerPoll must not be null.");
this.maxRowsPerPoll = maxRowsPerPoll;
}
@Override
protected void onInit() {
if (this.maxRowsPerPoll != null) {
Assert.notNull(poller, "If you want to set 'maxRowsPerPoll', then you must provide a 'selectQuery'.");
poller.setMaxRowsPerPoll(this.maxRowsPerPoll);
}
handler.afterPropertiesSet();
}
@Override

12
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.
@@ -23,8 +23,10 @@ import org.w3c.dom.Element;
/**
* @author Dave Syer
* @since 2.0
* @author Gunnar Hillert
*
* @since 2.0
*
*/
public class JdbcOutboundGatewayParser extends AbstractConsumerEndpointParser {
@@ -75,7 +77,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");

112
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>
@@ -585,14 +595,20 @@
<xsd:element name="sql-parameter-definition" minOccurs="0"
maxOccurs="unbounded" type="sqlParameterDefinitionType">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
For fully supported database these parameters
need not be declared as for those database the
type information can be retrieved from the
JDBC Metadata.
<xsd:documentation><![CDATA[
If you are using a database that is fully supported,
you typically don't have to specify the Stored Procedure
parameter definitions using the 'sql-parameter-definition'
attribute.
Instead, those parameters can be automatically derived
from the JDBC Meta-data. However, if you are using
databases that are not fully supported or if you like
to provide customized parameter definitions, you can
set those parameters explicitly. See also the
'ignore-column-meta-data' attribute.
Fully Supported Databases (Stored Procedures):
Fully Supported Databases (Stored Procedures):
* Apache Derby
* DB2
@@ -602,16 +618,14 @@
* PostgreSQL
* Sybase
Fully Supported Databases (Functions)
Fully Supported Databases (Functions)
* MySQL
* Microsoft SQL Server
* Oracle
* PostgreSQL
If you use a database not listed above, you
MUST provide Sql Parameter Definitions.
]]>
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:element>
@@ -721,35 +735,39 @@
<xsd:element name="sql-parameter-definition" minOccurs="0"
maxOccurs="unbounded" type="sqlParameterDefinitionType">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
For fully supported database these parameters
generally need not be declared as for those
databases the type information can be
retrieved from the JDBC Metadata.
Fully Supported Databases (Stored Procedures):
* Apache Derby
* DB2
* MySQL
* Microsoft SQL Server
* Oracle
* PostgreSQL
* Sybase
Fully Supported Databases (Functions)
* MySQL
* Microsoft SQL Server
* Oracle
* PostgreSQL
If you use a database not listed above, you
MUST provide Sql Parameter Definitions.
]]>
<xsd:documentation><![CDATA[
If you are using a database that is fully supported,
you typically don't have to specify the Stored Procedure
parameter definitions using the 'sql-parameter-definition'
attribute.
Instead, those parameters can be automatically derived
from the JDBC Meta-data. However, if you are using
databases that are not fully supported or if you like
to provide customized parameter definitions, you can
set those parameters explicitly. See also the
'ignore-column-meta-data' attribute.
Fully Supported Databases (Stored Procedures):
* Apache Derby
* DB2
* MySQL
* Microsoft SQL Server
* Oracle
* PostgreSQL
* Sybase
Fully Supported Databases (Functions)
* MySQL
* Microsoft SQL Server
* Oracle
* PostgreSQL
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:annotation>
</xsd:element>
<xsd:element name="parameter" minOccurs="0" maxOccurs="unbounded"
type="parameterSubElementType">

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

@@ -0,0 +1,75 @@
/*
* 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
public void testSetMaxRowsPerPollWithoutSelectQuery() {
DataSource dataSource = new EmbeddedDatabaseBuilder().build();
JdbcOutboundGateway jdbcOutboundGateway = new JdbcOutboundGateway(dataSource, "select * from DOES_NOT_EXIST");
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;
}
fail("Expected an IllegalArgumentException to be thrown.");
}
/**
* Test method for
* {@link org.springframework.integration.jdbc.JdbcOutboundGateway#setMaxRowsPerPoll(Integer)}.
*/
@Test
public void testSetMaxRowsPerPoll() {
DataSource dataSource = new EmbeddedDatabaseBuilder().build();
JdbcOutboundGateway jdbcOutboundGateway = new JdbcOutboundGateway(dataSource, "select * from DOES_NOT_EXIST");
try {
jdbcOutboundGateway.setMaxRowsPerPoll(null);
} catch (IllegalArgumentException e) {
assertEquals("MaxRowsPerPoll must not be null.", 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>