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

INT-1472, reveresed application of 'extractRequestPayload' and 'extractReplyPayload' attributes, changed javadocs, added test as well as minor modification to ChannelPublishingMessageListener to deal with RuntimeException (e.t., NotSerializable) during heneration of the JMS Reply

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-09-22 10:20:25 -04:00
parent c4d31af848
commit dee91f6871
6 changed files with 318 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -162,10 +162,10 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
/**
* Provide a {@link MessageConverter} implementation to use when
* converting between JMS Messages and Spring Integration Messages.
* If none is provided, a {@link DefaultMessageConverter} will
* be used and the {@link JmsHeaderMapper} instance provided to the
* {@link #setHeaderMapper(JmsHeaderMapper)} method will be included
* in the conversion process.
* If none is provided, a {@link SimpleMessageConverter} will
* be used.
*
* @param messageConverter
*/
public void setMessageConverter(MessageConverter messageConverter) {
this.messageConverter = messageConverter;
@@ -175,11 +175,6 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
* Provide a {@link JmsHeaderMapper} implementation to use when
* converting between JMS Messages and Spring Integration Messages.
* If none is provided, a {@link DefaultJmsHeaderMapper} will be used.
*
* <p>This property will be ignored if a {@link MessageConverter} is
* provided to the {@link #setMessageConverter(MessageConverter)} method.
* However, you may provide your own implementation of the delegating
* {@link DefaultMessageConverter} implementation.
*/
public void setHeaderMapper(JmsHeaderMapper headerMapper) {
this.headerMapper = headerMapper;
@@ -232,25 +227,34 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
if (this.extractReplyPayload){
replyResult = replyMessage.getPayload();
}
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
if (jmsReply.getJMSCorrelationID() == null) {
jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID());
}
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {
producer.send(jmsReply,
this.replyDeliveryMode, this.replyPriority, this.replyTimeToLive);
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
if (jmsReply.getJMSCorrelationID() == null) {
jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID());
}
else {
producer.send(jmsReply);
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {
producer.send(jmsReply,
this.replyDeliveryMode, this.replyPriority, this.replyTimeToLive);
}
else {
producer.send(jmsReply);
}
}
}
finally {
producer.close();
finally {
producer.close();
}
} catch (RuntimeException e) {
//e.printStackTrace();
logger.error("Problems generating JMS Reply Message from<6F>the: " + replyResult + "\n" +
" Typical couses is that Object from which the JMS Message is created or some members of its " +
"hierarchy are not Serializable", e);
throw e;
}
}
}

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

@@ -201,8 +201,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
* Spring Integration request Message into a JMS Message and for converting
* the JMS reply Messages back into Spring Integration Messages.
* <p>
* The default is a {@link DefaultMessageConverter} that delegates to
* a {@link SimpleMessageConverter}.
* The default is {@link SimpleMessageConverter}.
*/
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null");
@@ -212,37 +211,34 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
/**
* Provide a {@link JmsHeaderMapper} implementation for mapping the
* Spring Integration Message Headers to/from JMS Message properties.
*
* <p>This property will be ignored if a {@link MessageConverter} is
* provided to the {@link #setMessageConverter(MessageConverter)} method.
* However, you may provide your own implementation of the delegating
* {@link DefaultMessageConverter} implementation.
*/
public void setHeaderMapper(JmsHeaderMapper headerMapper) {
this.headerMapper = headerMapper;
}
/**
* This property will take effect if no custom {@link MessageConverter}
* has been provided to the {@link #setMessageConverter(MessageConverter)}
* method. In that case, a {@link DefaultMessageConverter} will be
* used by default, and this value will be passed along to that converter's
* 'extractIntegrationMessagePayload' property.
* This property describes how JMS Message should be generated from
* Spring Integration (SI) Message. If set to 'true', the base for JMS Message will be
* SI Message's payload, if set to 'false', then the entire SI Message will serve as
* a base for JMS Message creation.
*
* @see DefaultMessageConverter#setExtractIntegrationMessagePayload(boolean)
* Since JMS Message is created by the MessageConverter, this really manages what
* is sent to a {@link MessageConverter} - the entire SI Message or only its payload.
* <br>
* Default is 'true'
*
* @param extractRequestPayload
*/
public void setExtractRequestPayload(boolean extractRequestPayload) {
this.extractRequestPayload = extractRequestPayload;
}
/**
* This property will take effect if no custom {@link MessageConverter}
* has been provided to the {@link #setMessageConverter(MessageConverter)}
* method. In that case, a {@link DefaultMessageConverter} will be
* used by default, and this value will be passed along to that converter's
* 'extractJmsMessageBody' property.
* This property describes what to do with JMS Message after reply was received.
* If set to 'true', the base for SI Message will be JMS Reply Message's payload
* otherwise the entire JMS Message will become a payload of SI Message.
*
* @see DefaultMessageConverter#setExtractJmsMessageBody(boolean)
* @param extractReplyPayload
*/
public void setExtractReplyPayload(boolean extractReplyPayload) {
this.extractReplyPayload = extractReplyPayload;
@@ -305,7 +301,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
"failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms");
}
Object result = jmsReply;
if (this.extractRequestPayload) {
if (this.extractReplyPayload) {
result = this.messageConverter.fromMessage(jmsReply);
if (logger.isDebugEnabled()) {
logger.debug("converted JMS Message [" + jmsReply + "] to integration Message payload [" + result + "]");
@@ -328,7 +324,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
session = createSession(connection);
// convert to JMS Message
Object objectToSend = requestMessage;
if (this.extractReplyPayload){
if (this.extractRequestPayload){
objectToSend = requestMessage.getPayload();
}
javax.jms.Message jmsRequest = this.messageConverter.toMessage(objectToSend, session);

48
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests-context.xml Normal file
View File

@@ -0,0 +1,48 @@
<?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-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.0.xsd
http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:task="http://www.springframework.org/schema/task">
<int:channel id="outboundChannel"/>
<int:channel id="replyChannel">
<int:queue/>
</int:channel>
<int-jms:outbound-gateway id="outboundGateway"
request-channel="outboundChannel"
request-destination="extractRecRepQueue"
reply-channel="replyChannel"
receive-timeout="3000"
reply-timeout="3000"/>
<int-jms:inbound-gateway id="inboundGateway"
request-destination="extractRecRepQueue"
request-channel="jmsInputChannel"
request-timeout="3000"
reply-timeout="3000"/>
<int:channel id="jmsInputChannel"/>
<bean id="extractRecRepQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="request.queue.req.rep"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
</beans>

213
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests.java Normal file
View File

@@ -0,0 +1,213 @@
/**
*
*/
package org.springframework.integration.jms.config;
import static junit.framework.Assert.assertTrue;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.message.GenericMessage;
/**
* @author ozhurakousky
*
*/
public class ExtractRequestReplyPayloadTests {
ClassPathXmlApplicationContext applicationContext;
MessageChannel outboundChannel;
SubscribableChannel jmsInputChannel;
PollableChannel replyChannel;
@Before
public void prepare(){
ActiveMqTestUtils.prepare();
applicationContext = new ClassPathXmlApplicationContext("ExtractRequestReplyPayloadTests-context.xml", this.getClass());
outboundChannel = applicationContext.getBean("outboundChannel", MessageChannel.class);
jmsInputChannel = applicationContext.getBean("jmsInputChannel", SubscribableChannel.class);
replyChannel = applicationContext.getBean("replyChannel", PollableChannel.class);
}
@After
public void cleanup(){
applicationContext.destroy();
}
@Test
public void testOutboundInboundDefault(){
jmsInputChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
});
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof String);
}
@Test
public void testOutboundBothFalseInboundDefault(){
JmsOutboundGateway outboundGateway =
(JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler");
outboundGateway.setExtractRequestPayload(false);
outboundGateway.setExtractReplyPayload(false);
jmsInputChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
});
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof javax.jms.Message);
}
@Test(expected=MessageTimeoutException.class)
public void testOutboundDefaultInboundBothTrue(){
ChannelPublishingJmsMessageListener inboundGateway =
(ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")).
getPropertyValue("listener");
inboundGateway.setExtractReplyPayload(false);
inboundGateway.setExtractRequestPayload(false);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof javax.jms.Message);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
};
handler = spy(handler);
jmsInputChannel.subscribe(handler);
outboundChannel.send(new GenericMessage<String>("Hello"));
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
replyChannel.receive(1000);
}
@Test
public void testOutboundDefaultInboundReplyTrueRequestFalse(){
ChannelPublishingJmsMessageListener inboundGateway =
(ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")).
getPropertyValue("listener");
inboundGateway.setExtractReplyPayload(true);
inboundGateway.setExtractRequestPayload(false);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof javax.jms.Message);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
};
jmsInputChannel.subscribe(handler);
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof String);
}
@Test
public void testOutboundDefaultInboundReplyFalseRequestTrue(){
ChannelPublishingJmsMessageListener inboundGateway =
(ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")).
getPropertyValue("listener");
inboundGateway.setExtractReplyPayload(false);
inboundGateway.setExtractRequestPayload(true);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
};
jmsInputChannel.subscribe(handler);
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof String);
}
@Test
public void testOutboundRequestTrueReplyFalseInboundDefault(){
JmsOutboundGateway outboundGateway =
(JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler");
outboundGateway.setExtractRequestPayload(true);
outboundGateway.setExtractReplyPayload(false);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
};
jmsInputChannel.subscribe(handler);
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof javax.jms.Message);
}
@Test
public void testOutboundRequestFalseReplyTrueInboundDefault(){
JmsOutboundGateway outboundGateway =
(JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler");
outboundGateway.setExtractRequestPayload(false);
outboundGateway.setExtractReplyPayload(true);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
};
jmsInputChannel.subscribe(handler);
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof String);
}
@Test(expected=MessageTimeoutException.class)
public void testAllFalse(){
JmsOutboundGateway outboundGateway =
(JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler");
outboundGateway.setExtractRequestPayload(false);
outboundGateway.setExtractReplyPayload(false);
ChannelPublishingJmsMessageListener inboundGateway =
(ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")).
getPropertyValue("listener");
inboundGateway.setExtractReplyPayload(false);
inboundGateway.setExtractRequestPayload(false);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof javax.jms.Message);
MessagingTemplate template = new MessagingTemplate((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
};
jmsInputChannel.subscribe(handler);
outboundChannel.send(new GenericMessage<String>("Hello"));
Message<?> replyMessage = replyChannel.receive(1000);
assertTrue(replyMessage.getPayload() instanceof String);
}
}

0
spring-integration-jms/src/test/resources/.svnignore
View File

11
spring-integration-jms/src/test/resources/log4j.properties Normal file
View File

@@ -0,0 +1,11 @@
log4j.rootCategory=WARN, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n
log4j.category.org.springframework=WARN
# log4j.category.org.springframework.integration=DEBUG
# log4j.category.org.springframework.integration.jdbc=DEBUG
log4j.category.org.springframework.jms=DEBUG