diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcOutboundGateway.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcOutboundGateway.java index 5e3fa6ef5f..8ca248e6a7 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcOutboundGateway.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -26,7 +26,11 @@ import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; /** + * An {@link AbstractReplyProducingMessageHandler} implementation for performing + * RDBMS stored procedures which return results. + * * @author Gunnar Hillert + * @author Artem Bilan * * @since 2.1 */ @@ -36,17 +40,45 @@ public class StoredProcOutboundGateway extends AbstractReplyProducingMessageHand private volatile boolean expectSingleResult = false; + private boolean requiresReplyExplicitlySet; + /** * Constructor taking {@link StoredProcExecutor}. - * * @param storedProcExecutor Must not be null. - * */ public StoredProcOutboundGateway(StoredProcExecutor storedProcExecutor) { - Assert.notNull(storedProcExecutor, "storedProcExecutor must not be null."); this.executor = storedProcExecutor; + } + @Override + public void setRequiresReply(boolean requiresReply) { + super.setRequiresReply(requiresReply); + this.requiresReplyExplicitlySet = true; + } + + /** + * This parameter indicates that only one result object shall be returned from + * the Stored Procedure/Function Call. If set to {@code true}, a {@code resultMap} that contains + * only 1 element, will have that 1 element extracted and returned as payload. + *

If the {@code resultMap} contains more than 1 element and {@code expectSingleResult == true}, + * then a {@link MessagingException} is thrown. + *

Otherwise the complete {@code resultMap} is returned as the {@link Message} payload. + *

Important Note: Several databases such as H2 are not fully supported. + * The H2 database, for example, does not fully support the {@link CallableStatement} + * semantics and when executing function calls against H2, a result list is + * returned rather than a single value. + *

Therefore, even if you set {@code expectSingleResult = true}, you may end up with + * a collection being returned. + *

When set to {@code true}, a {@link #setRequiresReply(boolean)} is called + * with {@code true} as well, indicating that exactly single result is expected + * and {@code null} isn't appropriate value. + * A {@link org.springframework.integration.handler.ReplyRequiredException} is thrown + * in case of {@code null} result. + * @param expectSingleResult true if a single result is expected. + */ + public void setExpectSingleResult(boolean expectSingleResult) { + this.expectSingleResult = expectSingleResult; } @Override @@ -55,8 +87,15 @@ public class StoredProcOutboundGateway extends AbstractReplyProducingMessageHand } @Override - protected Object handleRequestMessage(Message requestMessage) { + protected void doInit() { + super.doInit(); + if (!this.requiresReplyExplicitlySet) { + setRequiresReply(this.expectSingleResult); + } + } + @Override + protected Object handleRequestMessage(Message requestMessage) { Map resultMap = this.executor.executeStoredProcedure(requestMessage); final Object payload; @@ -65,7 +104,6 @@ public class StoredProcOutboundGateway extends AbstractReplyProducingMessageHand return null; } else { - if (this.expectSingleResult && resultMap.size() == 1) { payload = resultMap.values().iterator().next(); } @@ -73,41 +111,16 @@ public class StoredProcOutboundGateway extends AbstractReplyProducingMessageHand throw new MessageHandlingException(requestMessage, "Stored Procedure/Function call returned more than " - + "1 result object and expectSingleResult was 'true'. "); - + + "1 result object and expectSingleResult was 'true'. "); } else { payload = resultMap; } - } - return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(requestMessage.getHeaders()).build(); - + return payload; } - /** - * This parameter indicates that only one result object shall be returned from - * the Stored Procedure/Function Call. If set to true, a resultMap that contains - * only 1 element, will have that 1 element extracted and returned as payload. - * - * If the resultMap contains more than 1 element and expectSingleResult is true, - * then a {@link MessagingException} is thrown. - * - * Otherwise the complete resultMap is returned as the {@link Message} payload. - * - * Important Note: Several databases such as H2 are not fully supported. - * The H2 database, for example, does not fully support the {@link CallableStatement} - * semantics and when executing function calls against H2, a result list is - * returned rather than a single value. - * - * Therefore, even if you set expectSingleResult = true, you may end up with - * a collection being returned. - * - * @param expectSingleResult true if a single result is expected. - */ - public void setExpectSingleResult(boolean expectSingleResult) { - this.expectSingleResult = expectSingleResult; - } + } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParser.java index 8731fdf3d3..8e301f0ba8 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParser.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -51,11 +51,11 @@ public class StoredProcOutboundGatewayParser extends AbstractConsumerEndpointPar IntegrationNamespaceUtils.setReferenceIfAttributeDefined(storedProcExecutorBuilder, element, "sql-parameter-source-factory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(storedProcExecutorBuilder, element, "skip-undeclared-results"); - final ManagedMap returningResultsetMap = + final ManagedMap returningResultSetMap = StoredProcParserUtils.getReturningResultsetBeanDefinitions(element, parserContext); - if (!returningResultsetMap.isEmpty()) { - storedProcExecutorBuilder.addPropertyValue("returningResultSetRowMappers", returningResultsetMap); + if (!returningResultSetMap.isEmpty()) { + storedProcExecutorBuilder.addPropertyValue("returningResultSetRowMappers", returningResultSetMap); } final AbstractBeanDefinition storedProcExecutorBuilderBeanDefinition = storedProcExecutorBuilder.getBeanDefinition(); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java index 205801db35..d32e23096e 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java @@ -656,7 +656,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto */ public void removeFromIdCache(String messageId) { if (logger.isDebugEnabled()) { - logger.debug("Removing Message Id:" + messageId); + logger.debug("Removing Message Id: " + messageId); } this.idCacheWriteLock.lock(); try { diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.3.xsd b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.3.xsd index e6464787a8..18198e10dd 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.3.xsd +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-4.3.xsd @@ -963,7 +963,7 @@ - + Specify whether this outbound gateway must return a non-null value. This value is diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java index 4c03033e4e..012569c3be 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests-context.xml index b8b97f3ad3..52e400a127 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests-context.xml @@ -2,17 +2,10 @@ + http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests.java index b5b1d686d7..b25d14dfb0 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.sql.CallableStatement; import java.util.Collection; @@ -38,6 +39,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.handler.ReplyRequiredException; import org.springframework.integration.jdbc.config.JdbcTypesEnum; import org.springframework.integration.jdbc.storedproc.User; import org.springframework.integration.support.MessageBuilder; @@ -149,6 +151,7 @@ public class StoredProcOutboundGatewayWithSpelIntegrationTests { @Test @Transactional public void testInt2865SqlReturnType() throws Exception { + Mockito.reset(this.clobSqlReturnType); Message testMessage = MessageBuilder.withPayload("TEST").setHeader("FOO", "BAR").build(); String messageId = testMessage.getHeaders().getId().toString(); String jsonMessage = new JsonOutboundMessageMapper().fromMessage(testMessage); @@ -168,6 +171,17 @@ public class StoredProcOutboundGatewayWithSpelIntegrationTests { Mockito.eq(2), Mockito.eq(JdbcTypesEnum.CLOB.getCode()), Mockito.eq((String) null)); } + @Test + public void testNoIllegalArgumentButRequiresReplyException() { + try { + this.getMessageChannel.send(new GenericMessage("foo")); + fail("ReplyRequiredException expected"); + } + catch (Exception e) { + assertThat(e, instanceOf(ReplyRequiredException.class)); + } + } + static class Counter { private final AtomicInteger count = new AtomicInteger(); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java index 9f08f077ef..cb8e7ca588 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.util.Collections; +import java.util.List; import java.util.Map; import javax.sql.DataSource; @@ -33,6 +34,7 @@ import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvi import org.springframework.integration.jdbc.JdbcMessageHandler; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.jdbc.core.ColumnMapRowMapper; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -117,8 +119,18 @@ public class JdbcMessageHandlerParserTests { MessageChannel target = context.getBean("target", MessageChannel.class); Message message = MessageBuilder.withPayload("foo").setHeader("business.key", "FOO").build(); target.send(message); - Thread.sleep(2000); - Map map = (context.getBean("jdbcTemplate", JdbcTemplate.class)).queryForMap("SELECT * from FOOW"); + + JdbcTemplate jdbcTemplate = context.getBean("jdbcTemplate", JdbcTemplate.class); + int n = 0; + List> result = jdbcTemplate.query("SELECT * from FOOW", new ColumnMapRowMapper()); + while (result.isEmpty() && n++ < 100) { + Thread.sleep(100); + result = jdbcTemplate.query("SELECT * from FOOW", new ColumnMapRowMapper()); + } + assertTrue(n < 100); + + assertEquals(1, result.size()); + Map map = result.get(0); assertEquals("Wrong id", "FOO", map.get("ID")); assertEquals("Wrong id", "foo", map.get("name")); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java index 63ee22c0e9..2203f6714e 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -50,6 +50,8 @@ import org.springframework.transaction.support.TransactionTemplate; /** * @author David Syer * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 * */ @@ -69,6 +71,7 @@ public class JdbcPollingChannelAdapterParserTests { public void testNoAutoStartupInboundChannelAdapter() { setUp("pollingNoAutoStartupJdbcInboundChannelAdapterTest.xml", getClass()); this.jdbcTemplate.update("insert into item values(1,'',2)"); + messagingTemplate.setReceiveTimeout(1); Message message = messagingTemplate.receive(); assertNull("Message found ", message); } @@ -94,8 +97,9 @@ public class JdbcPollingChannelAdapterParserTests { this.jdbcTemplate.update("insert into item values(1,'',2)"); Message message = messagingTemplate.receive(); assertNotNull(message); + messagingTemplate.setReceiveTimeout(1); message = messagingTemplate.receive(); - assertNull(messagingTemplate.receive()); + assertNull(message); } @Test @@ -104,8 +108,9 @@ public class JdbcPollingChannelAdapterParserTests { this.jdbcTemplate.update("insert into item values(1,'',2)"); Message message = messagingTemplate.receive(); assertNotNull(message); + messagingTemplate.setReceiveTimeout(1); message = messagingTemplate.receive(); - assertNull(messagingTemplate.receive()); + assertNull(message); } @Test diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParserTests.java index efa4c52022..2a718de927 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParserTests.java @@ -171,7 +171,7 @@ public class StoredProcOutboundGatewayParserTests { @SuppressWarnings("unchecked") @Test - public void testProcedurepParametersAreSet() throws Exception { + public void testProcedureParametersAreSet() throws Exception { setUp("storedProcOutboundGatewayParserTest.xml", getClass()); DirectFieldAccessor accessor = new DirectFieldAccessor(this.outboundGateway); @@ -200,7 +200,7 @@ public class StoredProcOutboundGatewayParserTests { assertEquals("kenny", parameter1.getValue()); assertEquals("Who killed Kenny?", parameter2.getValue()); assertNull(parameter3.getValue()); - assertEquals(Integer.valueOf(30), parameter4.getValue()); + assertEquals(30, parameter4.getValue()); assertNull(parameter1.getExpression()); assertNull(parameter2.getExpression()); @@ -273,11 +273,8 @@ public class StoredProcOutboundGatewayParserTests { assertEquals("SqlType is ", Types.INTEGER, parameter3.getSqlType()); assertEquals("SqlType is ", Types.VARCHAR, parameter4.getSqlType()); - assertTrue(parameter1 instanceof SqlParameter); assertTrue(parameter2 instanceof SqlOutParameter); assertTrue(parameter3 instanceof SqlInOutParameter); - assertTrue(parameter4 instanceof SqlParameter); - } @Test diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/advisedStoredProcOutboundGatewayParserTest.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/advisedStoredProcOutboundGatewayParserTest.xml index c9477ef1d1..ece9802693 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/advisedStoredProcOutboundGatewayParserTest.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/advisedStoredProcOutboundGatewayParserTest.xml @@ -1,39 +1,41 @@ - - + + - + - + - - - - - - - - - + + + + + + + + + - - - - + + + + + + - diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java index 586d814a76..4bab2e8f5c 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -101,6 +101,9 @@ public abstract class AbstractTxTimeoutMessageStoreTests { @Autowired protected PollableChannel priorityChannel; + @Autowired + protected PollableChannel afterTxChannel; + @Test public void test() throws InterruptedException { @@ -131,7 +134,10 @@ public abstract class AbstractTxTimeoutMessageStoreTests { Assert.assertTrue(String.format("Countdown latch did not count down from " + "%s to 0 in %sms.", maxMessages, maxWaitTime), testService.await(maxWaitTime)); - Thread.sleep(2000); + for (int i = 0; i < maxMessages; i++) { + Message afterTxMessage = this.afterTxChannel.receive(10000); + assertNotNull(afterTxMessage); + } Assert.assertEquals(Integer.valueOf(0), Integer.valueOf(jdbcChannelMessageStore.getSizeOfIdCache())); Assert.assertEquals(Integer.valueOf(maxMessages), Integer.valueOf(testService.getSeenMessages().size())); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java index 80fb1cede9..a609f7f727 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -69,7 +69,7 @@ public class TestService { if (this.threadSleep > 0) { try { - Thread.sleep(2000); + Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml index 260161dbbd..19436aa786 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml @@ -2,22 +2,20 @@ + http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> - - + + + + + +