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

INT-4202: Fix StoredProcOutboundGateway NPE

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4202

The `StoredProcOutboundGateway` uses `MessageBuilder` directly for procedure result without any conditions.
If procedure result is `null`, `MessageBuilder.withPayload` throws `java.lang.IllegalArgumentException: payload must not be null`.

* Since the super `AbstractReplyProducingMessageHandler` class takes care about `null` reply properly via its `requiresReply` property and really uses `MessageBuilder`
 for reply, just fix `StoredProcOutboundGateway` to return procedure result as is from the `handleRequestMessage()` implementation
* Fix some typos and code style
* Increase some JDBC tests performance changing `Thread.sleep()` solution to proper `PollableChannel.receive()` or iterations over expected result

**Cherry-pick to 4.3.x**

Update `@Copyright` to 2017

Make `StoredProcOutboundGateway` as `requiresReply = true` by default

Revert `setRequiresReply(true)` change

Specify `setRequiresReply(true)` only if `expectSingleResult == true` and `setRequiresReply()` hasn't been called explicitly
This commit is contained in:
Artem Bilan
2017-01-03 11:40:39 -05:00
committed by Gary Russell
parent ebdba7f75c
commit b7d7d76789
14 changed files with 144 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java
View File

@@ -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.
@@ -126,10 +126,10 @@ public class JdbcMessageStoreChannelIntegrationTests {
synchronized (storeLock) {
boolean result1 = input.send(new GenericMessage<String>("foo"), 500L);
boolean result1 = input.send(new GenericMessage<String>("foo"), 100L);
// This will time out because the transaction has not committed yet
try {
Service.await(3000);
Service.await(300);
fail("Expected timeout");
}
catch (Exception e) {

11
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests-context.xml
View File

@@ -2,17 +2,10 @@
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
xsi:schemaLocation="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/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:derby-stored-procedures-setup-context.xml"/>

16
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithSpelIntegrationTests.java
View File

@@ -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<String> 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<String>("foo"));
fail("ReplyRequiredException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(ReplyRequiredException.class));
}
}
static class Counter {
private final AtomicInteger count = new AtomicInteger();

18
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageHandlerParserTests.java
View File

@@ -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<String, Object> map = (context.getBean("jdbcTemplate", JdbcTemplate.class)).queryForMap("SELECT * from FOOW");
JdbcTemplate jdbcTemplate = context.getBean("jdbcTemplate", JdbcTemplate.class);
int n = 0;
List<Map<String, Object>> 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<String, Object> map = result.get(0);
assertEquals("Wrong id", "FOO", map.get("ID"));
assertEquals("Wrong id", "foo", map.get("name"));
}

11
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java
View File

@@ -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.
@@ -48,6 +48,8 @@ import org.springframework.transaction.support.TransactionTemplate;
/**
* @author David Syer
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*
*/
@@ -67,6 +69,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);
}
@@ -92,8 +95,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
@@ -102,8 +106,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

7
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcOutboundGatewayParserTests.java
View File

@@ -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

56
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/advisedStoredProcOutboundGatewayParserTest.xml
View File

@@ -1,39 +1,41 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.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/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="requestChannel" />
<int:channel id="replyChannel" />
<int:channel id="requestChannel"/>
<int:channel id="replyChannel"/>
<jdbc:embedded-database id="datasource" type="HSQL" />
<jdbc:embedded-database id="datasource" type="HSQL"/>
<int-jdbc:stored-proc-outbound-gateway
request-channel="requestChannel" stored-procedure-name="GET_PRIME_NUMBERS"
data-source="datasource" auto-startup="true" id="storedProcedureOutboundGateway"
ignore-column-meta-data="false" is-function="false"
skip-undeclared-results="false" order="2" reply-channel="replyChannel"
reply-timeout="555" return-value-required="false">
<int-jdbc:stored-proc-outbound-gateway
request-channel="requestChannel" stored-procedure-name="GET_PRIME_NUMBERS"
data-source="datasource" auto-startup="true" id="storedProcedureOutboundGateway"
ignore-column-meta-data="false" is-function="false"
skip-undeclared-results="false" order="2" reply-channel="replyChannel"
reply-timeout="555" return-value-required="false">
<int-jdbc:sql-parameter-definition name="username" direction="IN" type="VARCHAR" />
<int-jdbc:sql-parameter-definition name="password" direction="OUT" />
<int-jdbc:sql-parameter-definition name="age" direction="INOUT" type="INTEGER" scale="5" />
<int-jdbc:sql-parameter-definition name="description" />
<int-jdbc:parameter name="username" value="kenny" type="java.lang.String" />
<int-jdbc:parameter name="description" value="Who killed Kenny?" />
<int-jdbc:parameter name="password" expression="payload.username" />
<int-jdbc:parameter name="age" value="30" type="java.lang.Integer" />
<int-jdbc:returning-resultset name="out" row-mapper="org.springframework.integration.jdbc.storedproc.PrimeMapper"/>
<int-jdbc:sql-parameter-definition name="username" direction="IN" type="VARCHAR"/>
<int-jdbc:sql-parameter-definition name="password" direction="OUT"/>
<int-jdbc:sql-parameter-definition name="age" direction="INOUT" type="INTEGER" scale="5"/>
<int-jdbc:sql-parameter-definition name="description"/>
<int-jdbc:parameter name="username" value="kenny" type="java.lang.String"/>
<int-jdbc:parameter name="description" value="Who killed Kenny?"/>
<int-jdbc:parameter name="password" expression="payload.username"/>
<int-jdbc:parameter name="age" value="30" type="java.lang.Integer"/>
<int-jdbc:returning-resultset name="out"
row-mapper="org.springframework.integration.jdbc.storedproc.PrimeMapper"/>
<int-jdbc:request-handler-advice-chain>
<bean class="org.springframework.integration.jdbc.config.StoredProcOutboundGatewayParserTests$FooAdvice" />
</int-jdbc:request-handler-advice-chain>
</int-jdbc:stored-proc-outbound-gateway>
<int-jdbc:request-handler-advice-chain>
<bean class="org.springframework.integration.jdbc.config.StoredProcOutboundGatewayParserTests$FooAdvice"/>
</int-jdbc:request-handler-advice-chain>
</int-jdbc:stored-proc-outbound-gateway>
<int:poller default="true" fixed-rate="10000"/>
<int:poller default="true" fixed-rate="10000" />
</beans>

10
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
View File

@@ -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.
@@ -99,6 +99,9 @@ public abstract class AbstractTxTimeoutMessageStoreTests {
@Autowired
protected PollableChannel priorityChannel;
@Autowired
protected PollableChannel afterTxChannel;
@Test
public void test() throws InterruptedException {
@@ -129,7 +132,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()));

4
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TestService.java
View File

@@ -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();

18
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml
View File

@@ -2,22 +2,20 @@
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/task/spring-jdbc.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" />
<int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/>
<int:before-commit expression="@store.removeFromIdCache(headers.id.toString())"/>
<int:after-commit channel="afterTxChannel"/>
</int:transaction-synchronization-factory>
<int:channel id="afterTxChannel">
<int:queue/>
</int:channel>
<task:executor id="pool" pool-size="10" queue-capacity="10" rejection-policy="CALLER_RUNS" />
<bean id="store" class="org.springframework.integration.jdbc.store.JdbcChannelMessageStore">