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

INT-3313: Fix JMS-outbound requires-reply Logic

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3313

Previously, when `requires-reply="false"` on a `<jms:outbound-gateway/>`
a `null` reply caused a `MessageTimeoutException`, instead of quiet ending of flow.

Conflicts:

	spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java

Resolved.
This commit is contained in:
Artem Bilan
2014-03-05 18:56:45 +02:00
committed by Gary Russell
parent 1de46a8bdd
commit ee73515047
4 changed files with 84 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -142,6 +142,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private final Object lifeCycleMonitor = new Object();
private volatile boolean requiresReply;
/**
* Set whether message delivery should be persistent or non-persistent,
* specified as a boolean value ("true" or "false"). This will set the delivery
@@ -389,6 +391,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
this.useReplyContainer = useReplyContainer;
}
@Override
public void setRequiresReply(boolean requiresReply) {
super.setRequiresReply(requiresReply);
this.requiresReply = requiresReply;
}
private Destination determineRequestDestination(Message<?> message, Session session) throws JMSException {
if (this.requestDestination != null) {
return this.requestDestination;
@@ -617,8 +625,13 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
jmsReply = this.sendAndReceiveWithContainer(requestMessage);
}
if (jmsReply == null) {
throw new MessageTimeoutException(message,
"failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms");
if (this.requiresReply) {
throw new MessageTimeoutException(message,
"failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms");
}
else {
return null;
}
}
Object result = jmsReply;
if (this.extractReplyPayload) {

21
spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests-context.xml
View File

@@ -18,12 +18,33 @@
<jms:outbound-channel-adapter destination="testQueue"/>
</int:chain>
<int:channel id="repliesChannel">
<int:queue/>
</int:channel>
<int:chain input-channel="outboundGatewayChainChannel" output-channel="repliesChannel">
<jms:outbound-gateway request-destination="testQueue2" receive-timeout="1000" requires-reply="false"/>
</int:chain>
<jms:message-driven-channel-adapter destination="testQueue" channel="receiveChannel"/>
<jms:inbound-gateway request-channel="inboundJmsChannel" request-destination="testQueue2"/>
<bean id="jmsRequiresReplyFlag" class="java.util.concurrent.atomic.AtomicInteger"/>
<int:chain input-channel="inboundJmsChannel">
<int:service-activator expression="@jmsRequiresReplyFlag.getAndIncrement() % 2 == 0 ? payload : null"/>
</int:chain>
<bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="test.queue"/>
</bean>
<bean id="testQueue2" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="test.queue2"/>
</bean>
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
</bean>

43
spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java
View File

@@ -16,33 +16,58 @@
package org.springframework.integration.jms;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.support.MessageBuilder;
import static org.junit.Assert.*;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* //INT-2275
*
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class JmsOutboundInsideChainTests {
@Autowired
private MessageChannel outboundChainChannel;
@Autowired
private PollableChannel receiveChannel;
@Autowired
private MessageChannel outboundGatewayChainChannel;
@Autowired
private PollableChannel repliesChannel;
@Test
public void testJmsOutboundChannelInsideChain(){
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("JmsOutboundInsideChainTests-context.xml", getClass());
PollableChannel receiveChannel = context.getBean("receiveChannel", PollableChannel.class);
MessageChannel outboundChainChannel = context.getBean("outboundChainChannel", MessageChannel.class);
String testString = "test";
Message<String> shippedMessage = MessageBuilder.withPayload(testString).build();
outboundChainChannel.send(shippedMessage);
Message<?> receivedMessage = receiveChannel.receive();
this.outboundChainChannel.send(shippedMessage);
Message<?> receivedMessage = this.receiveChannel.receive(2000);
assertEquals(testString, receivedMessage.getPayload());
context.close();
}
@Test
public void testJmsOutboundGatewayRequiresReply(){
this.outboundGatewayChainChannel.send(MessageBuilder.withPayload("test").build());
assertNotNull(this.repliesChannel.receive(2000));
this.outboundGatewayChainChannel.send(MessageBuilder.withPayload("test").build());
assertNull(this.repliesChannel.receive(2000));
}
}