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.
This commit is contained in:
committed by
Artem Bilan
parent
4cca684f36
commit
fd35d43aba
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.");
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:simpleType>
|
||||
|
||||
@@ -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<String>("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<String>("Hello"));
|
||||
};
|
||||
this.jmsInputChannel.subscribe(handler);
|
||||
this.outboundChannel.send(new GenericMessage<String>("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<String>("Hello"));
|
||||
MessageHandler handler = echoInboundStringHandler();
|
||||
this.jmsInputChannel.subscribe(handler);
|
||||
this.outboundChannel.send(new GenericMessage<String>("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<String>("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<String>("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<String>("Hello"));
|
||||
Message<?> replyMessage = replyChannel.receive(1000);
|
||||
assertTrue(replyMessage.getPayload() instanceof String);
|
||||
MessageHandler handler = unwrapTextMessageAndEchoHandler();
|
||||
this.jmsInputChannel.subscribe(handler);
|
||||
this.outboundChannel.send(new GenericMessage<String>("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<String>("Hello"));
|
||||
Message<?> replyMessage = replyChannel.receive(1000);
|
||||
MessageHandler handler = echoInboundStringHandler();
|
||||
this.jmsInputChannel.subscribe(handler);
|
||||
this.outboundChannel.send(new GenericMessage<String>("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<String>("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<String>("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<String>("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<String>("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<String>("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<String>("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<String>(payload));
|
||||
}
|
||||
};
|
||||
return handler;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<Class<?>, Level> oldLevels = new HashMap<Class<?>, 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));
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -168,6 +168,15 @@ The `error-channel` now is used for the conversion errors, which have caused a t
|
||||
|
||||
See <<jms-message-driven-channel-adapter>> and <<jms-inbound-gateway>> 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 <<jms-message-driven-channel-adapter>> and <<jms-inbound-gateway>> for more information.
|
||||
|
||||
===== Shared Subscriptions
|
||||
|
||||
Namespace support for shared subscriptions (JMS 2.0) has been added to message-driven endpoints and the
|
||||
|
||||
Reference in New Issue
Block a user