diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 8dd2cbe3d2..00db8f3528 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -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. - * - *

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Ę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; } } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 12e3b629a5..2a986108dc 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -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. *

- * 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. - * - *

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. + *
+ * 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); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests-context.xml new file mode 100644 index 0000000000..8742a55743 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests-context.xml @@ -0,0 +1,48 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests.java new file mode 100644 index 0000000000..2c17f90a89 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExtractRequestReplyPayloadTests.java @@ -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("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("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("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("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("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("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("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("Hello")); + Message replyMessage = replyChannel.receive(1000); + assertTrue(replyMessage.getPayload() instanceof String); + } +} diff --git a/spring-integration-jms/src/test/resources/.svnignore b/spring-integration-jms/src/test/resources/.svnignore deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/spring-integration-jms/src/test/resources/log4j.properties b/spring-integration-jms/src/test/resources/log4j.properties new file mode 100644 index 0000000000..0d11d824ba --- /dev/null +++ b/spring-integration-jms/src/test/resources/log4j.properties @@ -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