From fd35d43aba7de42b67eb1ef74355f2507d6e5199 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 31 Jul 2015 14:37:31 -0400 Subject: [PATCH] INT-3784: Use acknowledge="transacted" by Default JIRA: https://jira.spring.io/browse/INT-3784 Also - suppress WARN log when priority mapping fails - rework extract payload tests to use a single context Make a new `JmsMessageDrivenEndpoint` as `private` because it makes sense only for XML `BeanDefinition` variant. --- .../ChannelPublishingJmsMessageListener.java | 3 + .../jms/DefaultJmsHeaderMapper.java | 3 +- .../jms/JmsMessageDrivenEndpoint.java | 48 ++- .../integration/jms/JmsOutboundGateway.java | 6 + .../JmsMessageDrivenEndpointParser.java | 9 +- .../integration/jms/util/JmsAdapterUtils.java | 21 +- .../jms/config/spring-integration-jms-4.2.xsd | 5 +- .../ExtractRequestReplyPayloadTests.java | 340 +++++++++++------- .../test/rule/Log4jLevelAdjuster.java | 11 +- src/reference/asciidoc/jms.adoc | 10 +- src/reference/asciidoc/whats-new.adoc | 9 + 11 files changed, 308 insertions(+), 157 deletions(-) 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 6056b2c2ad..9d7485f3ea 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 @@ -341,6 +341,9 @@ public class ChannelPublishingJmsMessageListener Message replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage); if (replyMessage != null) { Destination destination = this.getReplyDestination(jmsMessage, session); + if (logger.isDebugEnabled()) { + logger.debug("Reply destination: " + destination); + } if (destination != null) { // convert SI Message to JMS Message Object replyResult = replyMessage; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java index ba85dfa03b..9f5cfa2721 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java @@ -150,7 +150,8 @@ public class DefaultJmsHeaderMapper implements JmsHeaderMapper { jmsMessage.setObjectProperty(propertyName, value); } catch (Exception e) { - if (headerName.startsWith("JMSX")) { + if (headerName.startsWith("JMSX") + || headerName.equals(IntegrationMessageHeaderAccessor.PRIORITY)) { if (logger.isTraceEnabled()) { logger.trace("skipping reserved header, it cannot be set by client: " + headerName); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index 6aeb9d4598..54813058cc 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.jms.util.JmsAdapterUtils; import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.util.Assert; /** @@ -35,12 +36,33 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos private final AbstractMessageListenerContainer listenerContainer; + private final boolean externalContainer; + private final ChannelPublishingJmsMessageListener listener; - + private volatile String sessionAcknowledgeMode; - + + /** + * Construct an instance with an externally configured container. + * @param listenerContainer the container. + * @param listener the listener. + */ public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, ChannelPublishingJmsMessageListener listener) { + this(listenerContainer, listener, true); + } + + /** + * Construct an instance with an argument indicating whether the container's ack mode should + * be overridden with {@link #setSessionAcknowledgeMode(String) sessionAcknowledgeMode}, default + * 'transacted'. + * @param listenerContainer the container. + * @param listener the listener. + * @param externalContainer true if the container is externally configured and should not have its ackmode + * coerced when no sessionAcknowledgeMode was supplied. + */ + private JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, + ChannelPublishingJmsMessageListener listener, boolean externalContainer) { Assert.notNull(listenerContainer, "listener container must not be null"); Assert.notNull(listener, "listener must not be null"); if (logger.isWarnEnabled() && listenerContainer.getMessageListener() != null) { @@ -51,12 +73,20 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos this.listener = listener; this.listenerContainer = listenerContainer; setPhase(Integer.MAX_VALUE / 2); + this.externalContainer = externalContainer; } + /** + * Set the session acknowledge mode on the listener container. It will override the + * container setting even if an external container is provided. Defaults to null + * (won't change container) if an external container is provided or `transacted` when + * the framework creates an implicit {@link DefaultMessageListenerContainer}. + * @param sessionAcknowledgeMode the acknowledge mode. + */ public void setSessionAcknowledgeMode(String sessionAcknowledgeMode) { this.sessionAcknowledgeMode = sessionAcknowledgeMode; } - + @Override public String getComponentType() { return "jms:message-driven-channel-adapter"; @@ -68,7 +98,12 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos if (!this.listenerContainer.isActive()) { this.listenerContainer.afterPropertiesSet(); } - Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(this.sessionAcknowledgeMode); + String sessionAcknowledgeMode = this.sessionAcknowledgeMode; + if (sessionAcknowledgeMode == null && !this.externalContainer + && DefaultMessageListenerContainer.class.isAssignableFrom(this.listenerContainer.getClass())) { + sessionAcknowledgeMode = JmsAdapterUtils.SESSION_TRANSACTED_STRING; + } + Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAcknowledgeMode); if (acknowledgeMode != null) { if (acknowledgeMode.intValue() == JmsAdapterUtils.SESSION_TRANSACTED) { this.listenerContainer.setSessionTransacted(true); @@ -94,6 +129,7 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos this.listener.stop(); } + @Override public void destroy() throws Exception { if (this.isRunning()) { this.stop(); @@ -102,12 +138,14 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos } + @Override public int beforeShutdown() { this.stop(); return 0; } + @Override public int afterShutdown() { return 0; } 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 17d18c72ce..00122d63f5 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 @@ -770,6 +770,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp jmsRequest.setJMSReplyTo(replyTo); connection.start(); + if (logger.isDebugEnabled()) { + logger.debug("ReplyTo: " + replyTo); + } Integer priority = new IntegrationMessageHeaderAccessor(requestMessage).getPriority(); if (priority == null) { @@ -824,6 +827,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp replyTo = this.determineReplyDestination(requestMessage, session); jmsRequest.setJMSReplyTo(replyTo); connection.start(); + if (logger.isDebugEnabled()) { + logger.debug("ReplyTo: " + replyTo); + } Integer priority = new IntegrationMessageHeaderAccessor(requestMessage).getPriority(); if (priority == null) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 5e8104d59d..4af00e3e5a 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -63,6 +63,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE, "destination-resolver", "transaction-manager", "concurrent-consumers", "max-concurrent-consumers", + "acknowledge", "max-messages-per-task", "selector", "receive-timeout", "recovery-interval", "idle-consumer-limit", "idle-task-execution-limit", @@ -117,6 +118,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition String listenerBeanName = this.parseMessageListener(element, parserContext, builder.getRawBeanDefinition()); builder.addConstructorArgReference(containerBeanName); builder.addConstructorArgReference(listenerBeanName); + builder.addConstructorArgValue(hasExternalContainer(element)); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge", "sessionAcknowledgeMode"); @@ -126,7 +128,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition private String parseMessageListenerContainer(Element element, ParserContext parserContext, BeanDefinition adapterBeanDefinition) { String containerClass = element.getAttribute("container-class"); - if (element.hasAttribute("container")) { + if (hasExternalContainer(element)) { if (StringUtils.hasText(containerClass)) { parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element); } @@ -198,6 +200,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition return beanName; } + + private boolean hasExternalContainer(Element element) { + return element.hasAttribute("container"); + } + private String parseMessageListener(Element element, ParserContext parserContext, BeanDefinition adapterBeanDefinition) { BeanDefinitionBuilder builder = BeanDefinitionBuilder diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/util/JmsAdapterUtils.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/util/JmsAdapterUtils.java index 8bb5f9ef2b..6e760294ee 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/util/JmsAdapterUtils.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/util/JmsAdapterUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -19,11 +19,20 @@ import org.springframework.util.StringUtils; /** * @author Liujiong + * @author Gary Russell * @since 4.1 * */ public abstract class JmsAdapterUtils { - + + public static final String AUTO_ACKNOWLEDGE_STRING = "auto"; + + public static final String DUPS_OK_ACKNOWLEDGE_STRING = "dups-ok"; + + public static final String CLIENT_ACKNOWLEDGE_STRING = "client"; + + public static final String SESSION_TRANSACTED_STRING = "transacted"; + public static final int SESSION_TRANSACTED = 0; public static final int AUTO_ACKNOWLEDGE = 1; @@ -35,16 +44,16 @@ public abstract class JmsAdapterUtils { public static Integer parseAcknowledgeMode(String acknowledge) { if (StringUtils.hasText(acknowledge)) { int acknowledgeMode = AUTO_ACKNOWLEDGE; - if ("transacted".equals(acknowledge)) { + if (SESSION_TRANSACTED_STRING.equals(acknowledge)) { acknowledgeMode = SESSION_TRANSACTED; } - else if ("dups-ok".equals(acknowledge)) { + else if (DUPS_OK_ACKNOWLEDGE_STRING.equals(acknowledge)) { acknowledgeMode = DUPS_OK_ACKNOWLEDGE; } - else if ("client".equals(acknowledge)) { + else if (CLIENT_ACKNOWLEDGE_STRING.equals(acknowledge)) { acknowledgeMode = CLIENT_ACKNOWLEDGE; } - else if (!"auto".equals(acknowledge)) { + else if (!AUTO_ACKNOWLEDGE_STRING.equals(acknowledge)) { throw new IllegalStateException("Invalid JMS 'acknowledge' setting: " + "only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported."); } diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd index 7f20ced2d5..e249f453c8 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd @@ -1317,7 +1317,10 @@ The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted". The latter effectively activates a locally transacted Session. 'transacted' is not allowed on the inbound-channel-adapter; use 'session-transacted' instead. - acknowlege="transacted" is used on the message-driven-channel-adapter. + acknowlege="transacted" is used on the message-driven-channel-adapter and inbound gateway. + Defaults to "transacted" when an implicit message listener container is configured. + Not allowed when using an externally configured listener container; configure the container + instead. ]]> 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 index ebfc8a2ab1..734f5b4534 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. You may obtain a copy of the License at @@ -12,184 +12,236 @@ */ package org.springframework.integration.jms.config; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; +import static org.junit.Assert.fail; -import org.junit.After; -import org.junit.Before; +import java.util.concurrent.atomic.AtomicBoolean; + +import javax.jms.JMSException; + +import org.apache.log4j.Level; +import org.junit.Rule; import org.junit.Test; -import org.mockito.Mockito; -import org.springframework.beans.DirectFieldAccessor; -import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.MessageTimeoutException; +import org.junit.rules.TestName; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.jms.ChannelPublishingJmsMessageListener; import org.springframework.integration.jms.JmsOutboundGateway; -import org.springframework.messaging.support.GenericMessage; +import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; -import org.springframework.integration.core.MessagingTemplate; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author ozhurakousky * @author Gunnar Hillert + * @author Gary Russell * */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class ExtractRequestReplyPayloadTests { - ClassPathXmlApplicationContext applicationContext; + + @Rule + public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, "org.springframework.integration", + "org.springframework.jms"); + + @Rule + public TestName testName = new TestName(); + + @Autowired + ApplicationContext applicationContext; + + @Autowired MessageChannel outboundChannel; + + @Autowired SubscribableChannel jmsInputChannel; + + @Autowired 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(); - } + + @Autowired + JmsOutboundGateway outboundGateway; + + @Autowired + ChannelPublishingJmsMessageListener inboundGateway; @Test public void testOutboundInboundDefault(){ - jmsInputChannel.subscribe(new MessageHandler() { + this.outboundGateway.setExtractRequestPayload(true); + this.outboundGateway.setExtractReplyPayload(true); + + this.inboundGateway.setExtractReplyPayload(true); + this.inboundGateway.setExtractRequestPayload(true); + + MessageHandler handler = echoInboundStringHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + + Message replyMessage = this.replyChannel.receive(10000); + assertTrue(replyMessage.getPayload() instanceof String); + this.jmsInputChannel.unsubscribe(handler); + } + + @Test + public void testOutboundInboundDefaultIsTx(){ + this.outboundGateway.setExtractRequestPayload(true); + this.outboundGateway.setExtractReplyPayload(true); + + this.inboundGateway.setExtractReplyPayload(true); + this.inboundGateway.setExtractRequestPayload(true); + + final AtomicBoolean failOnce = new AtomicBoolean(); + MessageHandler handler = new MessageHandler() { + @Override public void handleMessage(Message message) throws MessagingException { assertTrue(message.getPayload() instanceof String); + if (failOnce.compareAndSet(false, true)) { + throw new RuntimeException("test tx"); + } MessagingTemplate template = new MessagingTemplate(); template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel()); template.send(message); } - }); - outboundChannel.send(new GenericMessage("Hello")); + }; + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); - Message replyMessage = replyChannel.receive(1000); + Message replyMessage = this.replyChannel.receive(10000); assertTrue(replyMessage.getPayload() instanceof String); + this.jmsInputChannel.unsubscribe(handler); } @Test public void testOutboundBothFalseInboundDefault(){ + this.outboundGateway.setExtractRequestPayload(false); + this.outboundGateway.setExtractReplyPayload(false); - JmsOutboundGateway outboundGateway = - (JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler"); - outboundGateway.setExtractRequestPayload(false); - outboundGateway.setExtractReplyPayload(false); + this.inboundGateway.setExtractReplyPayload(true); + this.inboundGateway.setExtractRequestPayload(true); - jmsInputChannel.subscribe(new MessageHandler() { - public void handleMessage(Message message) throws MessagingException { - assertTrue(message.getPayload() instanceof String); - MessagingTemplate template = new MessagingTemplate(); - template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel()); - template.send(message); - } - }); - outboundChannel.send(new GenericMessage("Hello")); + MessageHandler handler = echoInboundStringHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); - Message replyMessage = replyChannel.receive(1000); - assertTrue(replyMessage.getPayload() instanceof javax.jms.Message); + Message replyMessage = this.replyChannel.receive(10000); + assertThat(replyMessage.getPayload(), instanceOf(javax.jms.TextMessage.class)); + this.jmsInputChannel.unsubscribe(handler); } - @Test(expected=MessageTimeoutException.class) - public void testOutboundDefaultInboundBothTrue(){ - ChannelPublishingJmsMessageListener inboundGateway = - (ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")). - getPropertyValue("listener"); - inboundGateway.setExtractReplyPayload(false); - inboundGateway.setExtractRequestPayload(false); + @Test + public void testOutboundDefaultInboundBothFalse() throws Exception{ + this.outboundGateway.setExtractRequestPayload(true); + this.outboundGateway.setExtractReplyPayload(true); - MessageHandler handler = new MessageHandler() { - public void handleMessage(Message message) throws MessagingException { - assertTrue(message.getPayload() instanceof javax.jms.Message); - MessagingTemplate template = new MessagingTemplate(); - template.setDefaultDestination((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); + this.inboundGateway.setExtractReplyPayload(false); + this.inboundGateway.setExtractRequestPayload(false); + + MessageHandler handler = unwrapTextMessageAndEchoHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + Message replyMessage = this.replyChannel.receive(10000); + assertThat(replyMessage.getPayload(), instanceOf(String.class)); + this.jmsInputChannel.unsubscribe(handler); } + @Test public void testOutboundDefaultInboundReplyTrueRequestFalse(){ + this.outboundGateway.setExtractRequestPayload(true); + this.outboundGateway.setExtractReplyPayload(true); - ChannelPublishingJmsMessageListener inboundGateway = - (ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")). - getPropertyValue("listener"); - inboundGateway.setExtractReplyPayload(true); - inboundGateway.setExtractRequestPayload(false); + this.inboundGateway.setExtractReplyPayload(true); + this.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(); - template.setDefaultDestination((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); + MessageHandler handler = unwrapTextMessageAndEchoHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + Message replyMessage = this.replyChannel.receive(10000); + assertThat(replyMessage.getPayload(), instanceOf(String.class)); + this.jmsInputChannel.unsubscribe(handler); } + @Test public void testOutboundDefaultInboundReplyFalseRequestTrue(){ + this.outboundGateway.setExtractRequestPayload(true); + this.outboundGateway.setExtractReplyPayload(true); - ChannelPublishingJmsMessageListener inboundGateway = - (ChannelPublishingJmsMessageListener)new DirectFieldAccessor(applicationContext.getBean("inboundGateway")). - getPropertyValue("listener"); - inboundGateway.setExtractReplyPayload(false); - inboundGateway.setExtractRequestPayload(true); + this.inboundGateway.setExtractReplyPayload(false); + this.inboundGateway.setExtractRequestPayload(true); - MessageHandler handler = new MessageHandler() { - public void handleMessage(Message message) throws MessagingException { - assertTrue(message.getPayload() instanceof String); - MessagingTemplate template = new MessagingTemplate(); - template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel()); - template.send(message); - } - }; - jmsInputChannel.subscribe(handler); - outboundChannel.send(new GenericMessage("Hello")); - Message replyMessage = replyChannel.receive(1000); + MessageHandler handler = echoInboundStringHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + Message replyMessage = this.replyChannel.receive(10000); assertTrue(replyMessage.getPayload() instanceof String); + this.jmsInputChannel.unsubscribe(handler); } + @Test public void testOutboundRequestTrueReplyFalseInboundDefault(){ - JmsOutboundGateway outboundGateway = - (JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler"); - outboundGateway.setExtractRequestPayload(true); - outboundGateway.setExtractReplyPayload(false); + this.outboundGateway.setExtractRequestPayload(true); + this.outboundGateway.setExtractReplyPayload(false); - MessageHandler handler = new MessageHandler() { - public void handleMessage(Message message) throws MessagingException { - assertTrue(message.getPayload() instanceof String); - MessagingTemplate template = new MessagingTemplate(); - template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel()); - template.send(message); - } - }; - jmsInputChannel.subscribe(handler); - outboundChannel.send(new GenericMessage("Hello")); - Message replyMessage = replyChannel.receive(1000); + this.inboundGateway.setExtractReplyPayload(true); + this.inboundGateway.setExtractRequestPayload(true); + + MessageHandler handler = echoInboundStringHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + Message replyMessage = this.replyChannel.receive(10000); assertTrue(replyMessage.getPayload() instanceof javax.jms.Message); + this.jmsInputChannel.unsubscribe(handler); } + @Test public void testOutboundRequestFalseReplyTrueInboundDefault(){ - JmsOutboundGateway outboundGateway = - (JmsOutboundGateway) new DirectFieldAccessor(applicationContext.getBean("outboundGateway")).getPropertyValue("handler"); - outboundGateway.setExtractRequestPayload(false); - outboundGateway.setExtractReplyPayload(true); + this.outboundGateway.setExtractRequestPayload(false); + this.outboundGateway.setExtractReplyPayload(true); + this.inboundGateway.setExtractReplyPayload(true); + this.inboundGateway.setExtractRequestPayload(true); + + MessageHandler handler = echoInboundStringHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + Message replyMessage = this.replyChannel.receive(10000); + assertThat(replyMessage.getPayload(), instanceOf(String.class)); + this.jmsInputChannel.unsubscribe(handler); + } + + @Test + public void testAllFalse() throws Exception{ + this.outboundGateway.setExtractRequestPayload(false); + this.outboundGateway.setExtractReplyPayload(false); + + this.inboundGateway.setExtractReplyPayload(false); + this.inboundGateway.setExtractRequestPayload(false); + + MessageHandler handler = unwrapObjectMessageAndEchoHandler(); + this.jmsInputChannel.subscribe(handler); + this.outboundChannel.send(new GenericMessage("Hello " + this.testName.getMethodName())); + Message replyMessage = this.replyChannel.receive(10000); + assertThat(replyMessage.getPayload(), instanceOf(javax.jms.Message.class)); + this.jmsInputChannel.unsubscribe(handler); + } + + private MessageHandler echoInboundStringHandler() { MessageHandler handler = new MessageHandler() { + @Override public void handleMessage(Message message) throws MessagingException { assertTrue(message.getPayload() instanceof String); MessagingTemplate template = new MessagingTemplate(); @@ -197,35 +249,47 @@ public class ExtractRequestReplyPayloadTests { template.send(message); } }; - jmsInputChannel.subscribe(handler); - outboundChannel.send(new GenericMessage("Hello")); - Message replyMessage = replyChannel.receive(1000); - assertTrue(replyMessage.getPayload() instanceof String); + return handler; } - @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); + private MessageHandler unwrapObjectMessageAndEchoHandler() { MessageHandler handler = new MessageHandler() { + @Override public void handleMessage(Message message) throws MessagingException { - assertTrue(message.getPayload() instanceof javax.jms.Message); + assertThat(message.getPayload(), instanceOf(javax.jms.ObjectMessage.class)); MessagingTemplate template = new MessagingTemplate(); template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel()); - template.send(message); + Message origMessage = null; + try { + origMessage = (Message) ((javax.jms.ObjectMessage) message.getPayload()).getObject(); + } + catch (JMSException e) { + fail("failed to deserialize message"); + } + template.send(origMessage); } }; - jmsInputChannel.subscribe(handler); - outboundChannel.send(new GenericMessage("Hello")); - Message replyMessage = replyChannel.receive(1000); - assertTrue(replyMessage.getPayload() instanceof String); + return handler; } + + private MessageHandler unwrapTextMessageAndEchoHandler() { + MessageHandler handler = new MessageHandler() { + @Override + public void handleMessage(Message message) throws MessagingException { + assertThat(message.getPayload(), instanceOf(javax.jms.TextMessage.class)); + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel()); + String payload = null; + try { + payload = ((javax.jms.TextMessage) message.getPayload()).getText(); + } + catch (JMSException e) { + fail("failed to deserialize message"); + } + template.send(new GenericMessage(payload)); + } + }; + return handler; + } + } diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/rule/Log4jLevelAdjuster.java b/spring-integration-test/src/main/java/org/springframework/integration/test/rule/Log4jLevelAdjuster.java index bb8f7128b0..721a5beb06 100755 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/rule/Log4jLevelAdjuster.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/rule/Log4jLevelAdjuster.java @@ -61,11 +61,13 @@ public class Log4jLevelAdjuster implements MethodRule { } @Override - public Statement apply(final Statement base, FrameworkMethod method, Object target) { + public Statement apply(final Statement base, final FrameworkMethod method, Object target) { return new Statement() { @Override public void evaluate() throws Throwable { - logger.debug("Overriding log level setting for: " + Arrays.asList(classes)); + logger.debug("++++++++++++++++++++++++++++ " + + "Overriding log level setting for: " + Arrays.asList(classes) + " for test " + + method.getName()); Map, Level> oldLevels = new HashMap, Level>(); for (Class cls : classes) { oldLevels.put(cls, LogManager.getLogger(cls).getEffectiveLevel()); @@ -80,8 +82,9 @@ public class Log4jLevelAdjuster implements MethodRule { base.evaluate(); } finally { - logger.debug("Restoring log level setting for: " + Arrays.asList(classes) + " and " - + Arrays.asList(categories)); + logger.debug("++++++++++++++++++++++++++++ " + + "Restoring log level setting for: " + Arrays.asList(classes) + " and " + + Arrays.asList(categories) + " for test " + method.getName()); // raw Class type used to avoid http://bugs.sun.com/view_bug.do?bug_id=6682380 for (@SuppressWarnings("rawtypes") Class cls : classes) { LogManager.getLogger(cls).setLevel(oldLevels.get(cls)); diff --git a/src/reference/asciidoc/jms.adoc b/src/reference/asciidoc/jms.adoc index 0977744cd5..65c2171cd1 100644 --- a/src/reference/asciidoc/jms.adoc +++ b/src/reference/asciidoc/jms.adoc @@ -65,7 +65,7 @@ In earlier versions, you had to inject a `JmsTemplate` with `sessionTransacted` Note, however, that setting `session-transacted` to `true` has little value because the transaction is committed immediately after the `receive()` and before the message is sent to the `channel`, If you want the entire flow to be transactional (for example if there is a downstream outbound channel adapter), you must use a `transactional` poller, with a `JmsTransactionManager`. -Or, consider using a `jms-message-driven-channel-adapter` with `acknowledge` set to `transacted`. +Or, consider using a `jms-message-driven-channel-adapter` with `acknowledge` set to `transacted` (the default). [[jms-message-driven-channel-adapter]] === Message-Driven Channel Adapter @@ -89,6 +89,10 @@ If you have a custom listener container implementation (usually a subclass of `D In that case, the attributes on the adapter are transferred to an instance of your custom container. ===== +IMPORTANT: Starting with _version 4.2_, the default `acknowledge` mode is `transacted`, unless an external +container is provided, in which case the container should be configured as needed. +It is recommended to use `transacted` with the `DefaultMessageListenerContainer` to avoid message loss. + The 'extract-payload' property has the same effect as described above, and once again its default value is 'true'. The poller sub-element is not applicable for a message-driven Channel Adapter, as it will be actively invoked. For most usage scenarios, the message-driven approach is better since the Messages will be passed along to the `MessageChannel` as soon as they are received from the underlying JMS consumer. @@ -206,6 +210,10 @@ for a durable subscription, `subscription-shared` for a shared subscription (req has been available since _version 4.2_). Use `subscription-name` to name the subscription. +IMPORTANT: Starting with _version 4.2_, the default `acknowledge` mode is `transacted`, unless an external +container is provided, in which case the container should be configured as needed. +It is recommended to use `transacted` with the `DefaultMessageListenerContainer` to avoid message loss. + [[jms-outbound-gateway]] === Outbound Gateway diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 288ed33561..126a8436e3 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -168,6 +168,15 @@ The `error-channel` now is used for the conversion errors, which have caused a t See <> and <> for more information. +===== Default Acknowledge Mode + +When using an implicitly defined `DefaultMessageListenerContainer`, the default `acknowledge` is now `transacted`. +`transacted` is recommended when using this container, to avoid message loss. +This default now applies to the message-driven inbound adapter and the inbound gateway, it was already the +default for jms-backed channels. + +See <> and <> for more information. + ===== Shared Subscriptions Namespace support for shared subscriptions (JMS 2.0) has been added to message-driven endpoints and the