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 index 638d1de4a8..e8fa488f83 100644 --- 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 @@ -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 zero. + * + * 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 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 index 07fdd202d3..b9f9a30c6e 100644 --- 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 @@ -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"); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.1.xsd b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.1.xsd index 5a59675f7a..5f8d484cd6 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.1.xsd +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.1.xsd @@ -171,13 +171,11 @@ - - - Limits the number of rows extracted per query (otherwise all rows - are extracted into the - outgoing message). - - + + Limits the number of rows extracted per query (otherwise all rows + are extracted into the + outgoing message). + @@ -361,6 +359,18 @@ + + + + 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. + + + @@ -585,14 +595,20 @@ - - + + ]]> @@ -721,35 +735,39 @@ - - + - + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcOutboundGatewayTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcOutboundGatewayTests.java new file mode 100644 index 0000000000..1bfe6658cf --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcOutboundGatewayTests.java @@ -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."); + + } +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundAdapterWithPollerTest-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundAdapterWithPollerTest-context.xml index 898cd25a1b..cad8e41093 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundAdapterWithPollerTest-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundAdapterWithPollerTest-context.xml @@ -9,11 +9,11 @@ xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc" xmlns:jdbc="http://www.springframework.org/schema/jdbc"> - + - + @@ -24,15 +24,15 @@ - - + + - + 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 index ff1c9cd3a5..f46b8ea18e 100644 --- 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 @@ -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 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) { diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPoller2Test-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPoller2Test-context.xml new file mode 100644 index 0000000000..15043e2eda --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPoller2Test-context.xml @@ -0,0 +1,40 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPollerTest-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPollerTest-context.xml index 8fb99cb16a..32e3ae0673 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPollerTest-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayWithPollerTest-context.xml @@ -9,11 +9,11 @@ xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc" xmlns:jdbc="http://www.springframework.org/schema/jdbc"> - + - + @@ -22,18 +22,17 @@ request-channel="target" reply-channel="output" data-source="dataSource" auto-startup="true" reply-timeout="444"> + + + + + + - - - - - - - - +