diff --git a/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java b/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java index 811709cd53..888f8e6b2b 100644 --- a/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java +++ b/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java @@ -52,6 +52,8 @@ public abstract class AbstractJmsListenerContainerFactoryThis setting primarily indicates what type of destination to resolve + * if dynamic destinations are enabled. + * @param replyPubSubDomain "true" for the Publish/Subscribe domain ({@link javax.jms.Topic Topics}), + * "false" for the Point-to-Point domain ({@link javax.jms.Queue Queues}) + * @see #setDestinationResolver + */ + public void setReplyPubSubDomain(boolean replyPubSubDomain) { + this.replyPubSubDomain = replyPubSubDomain; + } + + /** + * Return whether the Publish/Subscribe domain ({@link javax.jms.Topic Topics}) is used + * for replies. Otherwise, the Point-to-Point domain ({@link javax.jms.Queue Queues}) is + * used. + */ + @Override + public boolean isReplyPubSubDomain() { + if (this.replyPubSubDomain != null) { + return replyPubSubDomain; + } + else { + return isPubSubDomain(); + } + } + /** * Set the {@link MessageConverter} strategy for converting JMS Messages. * @since 4.1 diff --git a/spring-jms/src/main/java/org/springframework/jms/listener/MessageListenerContainer.java b/spring-jms/src/main/java/org/springframework/jms/listener/MessageListenerContainer.java index 9d77fbc57d..920c9d4029 100644 --- a/spring-jms/src/main/java/org/springframework/jms/listener/MessageListenerContainer.java +++ b/spring-jms/src/main/java/org/springframework/jms/listener/MessageListenerContainer.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. @@ -47,4 +47,12 @@ public interface MessageListenerContainer extends SmartLifecycle { */ boolean isPubSubDomain(); + /** + * Return whether the reply destination uses Publish/Subscribe domain + * ({@link javax.jms.Topic Topics}). Otherwise, the Point-to-Point domain + * ({@link javax.jms.Queue Queues}) is used. + *

By default, the value is identical to {@link #isPubSubDomain()}. + */ + boolean isReplyPubSubDomain(); + } diff --git a/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsActivationSpecConfig.java b/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsActivationSpecConfig.java index fc2fdf8551..6a6fde3b48 100644 --- a/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsActivationSpecConfig.java +++ b/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsActivationSpecConfig.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. @@ -46,6 +46,8 @@ public class JmsActivationSpecConfig { private boolean pubSubDomain = false; + private Boolean replyPubSubDomain; + private boolean subscriptionDurable = false; private boolean subscriptionShared = false; @@ -81,6 +83,19 @@ public class JmsActivationSpecConfig { return this.pubSubDomain; } + public void setReplyPubSubDomain(boolean replyPubSubDomain) { + this.replyPubSubDomain = replyPubSubDomain; + } + + public boolean isReplyPubSubDomain() { + if (this.replyPubSubDomain != null) { + return this.replyPubSubDomain; + } + else { + return isPubSubDomain(); + } + } + public void setSubscriptionDurable(boolean subscriptionDurable) { this.subscriptionDurable = subscriptionDurable; if (subscriptionDurable) { diff --git a/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManager.java b/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManager.java index a82524b387..774dd75e90 100644 --- a/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManager.java +++ b/spring-jms/src/main/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManager.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. @@ -200,4 +200,13 @@ public class JmsMessageEndpointManager extends GenericMessageEndpointManager throw new IllegalStateException("Could not determine pubSubDomain - no activation spec config is set"); } + @Override + public boolean isReplyPubSubDomain() { + JmsActivationSpecConfig config = getActivationSpecConfig(); + if (config != null) { + return config.isReplyPubSubDomain(); + } + throw new IllegalStateException("Could not determine reply pubSubDomain - no activation spec config is set"); + } + } diff --git a/spring-jms/src/main/resources/META-INF/spring.schemas b/spring-jms/src/main/resources/META-INF/spring.schemas index c7c42d4642..a6ee238a4a 100644 --- a/spring-jms/src/main/resources/META-INF/spring.schemas +++ b/spring-jms/src/main/resources/META-INF/spring.schemas @@ -4,4 +4,5 @@ http\://www.springframework.org/schema/jms/spring-jms-3.1.xsd=org/springframewor http\://www.springframework.org/schema/jms/spring-jms-3.2.xsd=org/springframework/jms/config/spring-jms-3.2.xsd http\://www.springframework.org/schema/jms/spring-jms-4.0.xsd=org/springframework/jms/config/spring-jms-4.0.xsd http\://www.springframework.org/schema/jms/spring-jms-4.1.xsd=org/springframework/jms/config/spring-jms-4.1.xsd -http\://www.springframework.org/schema/jms/spring-jms.xsd=org/springframework/jms/config/spring-jms-4.1.xsd +http\://www.springframework.org/schema/jms/spring-jms-4.2.xsd=org/springframework/jms/config/spring-jms-4.2.xsd +http\://www.springframework.org/schema/jms/spring-jms.xsd=org/springframework/jms/config/spring-jms-4.2.xsd diff --git a/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.2.xsd b/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.2.xsd new file mode 100644 index 0000000000..5122b3c44c --- /dev/null +++ b/spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.2.xsd @@ -0,0 +1,638 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java b/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java index 81afa3b2c7..9430491ff8 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.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. @@ -155,6 +155,7 @@ public class JmsListenerContainerFactoryTests { factory.setSessionTransacted(true); factory.setSessionAcknowledgeMode(Session.DUPS_OK_ACKNOWLEDGE); factory.setPubSubDomain(true); + factory.setReplyPubSubDomain(true); factory.setSubscriptionDurable(true); factory.setClientId("client-1234"); factory.setAutoStartup(false); @@ -167,6 +168,7 @@ public class JmsListenerContainerFactoryTests { assertEquals(true, container.isSessionTransacted()); assertEquals(Session.DUPS_OK_ACKNOWLEDGE, container.getSessionAcknowledgeMode()); assertEquals(true, container.isPubSubDomain()); + assertEquals(true, container.isReplyPubSubDomain()); assertEquals(true, container.isSubscriptionDurable()); assertEquals("client-1234", container.getClientId()); assertEquals(false, container.isAutoStartup()); diff --git a/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java b/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java index cfbf50c45c..26610dc0cb 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/JmsNamespaceHandlerTests.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. @@ -152,6 +152,8 @@ public class JmsNamespaceHandlerTests { context.getBean("testDestinationResolver"), container.getDestinationResolver()); assertEquals("explicit message converter not set", context.getBean("testMessageConverter"), container.getMessageConverter()); + assertEquals("Wrong pub/sub", true, container.isPubSubDomain()); + assertEquals("Wrong durable flag", true, container.isSubscriptionDurable()); assertEquals("wrong cache", DefaultMessageListenerContainer.CACHE_CONNECTION, container.getCacheLevel()); assertEquals("wrong concurrency", 3, container.getConcurrentConsumers()); assertEquals("wrong concurrency", 5, container.getMaxConcurrentConsumers()); @@ -173,6 +175,7 @@ public class JmsNamespaceHandlerTests { context.getBean("testResourceAdapter"),container.getResourceAdapter()); assertEquals("explicit message converter not set", context.getBean("testMessageConverter"), container.getActivationSpecConfig().getMessageConverter()); + assertEquals("Wrong pub/sub", true, container.isPubSubDomain()); assertEquals("wrong concurrency", 5, container.getActivationSpecConfig().getMaxConcurrency()); assertEquals("Wrong prefetch", 50, container.getActivationSpecConfig().getPrefetchSize()); assertEquals("Wrong phase", 77, container.getPhase()); @@ -249,6 +252,29 @@ public class JmsNamespaceHandlerTests { listener4.getActivationSpecConfig().getMaxConcurrency()); } + @Test + public void testResponseDestination() { + // JMS + DefaultMessageListenerContainer listener1 = this.context + .getBean("listener1", DefaultMessageListenerContainer.class); + DefaultMessageListenerContainer listener2 = this.context + .getBean("listener2", DefaultMessageListenerContainer.class); + assertEquals("Wrong destination type on listener1", true, listener1.isPubSubDomain()); + assertEquals("Wrong destination type on listener2", true, listener2.isPubSubDomain()); + assertEquals("Wrong response destination type on listener1", false, listener1.isReplyPubSubDomain()); + assertEquals("Wrong response destination type on listener2", false, listener2.isReplyPubSubDomain()); + + // JCA + JmsMessageEndpointManager listener3 = this.context + .getBean("listener3", JmsMessageEndpointManager.class); + JmsMessageEndpointManager listener4 = this.context + .getBean("listener4", JmsMessageEndpointManager.class); + assertEquals("Wrong destination type on listener3", true, listener3.isPubSubDomain()); + assertEquals("Wrong destination type on listener4", true, listener4.isPubSubDomain()); + assertEquals("Wrong response destination type on listener3", false, listener3.isReplyPubSubDomain()); + assertEquals("Wrong response destination type on listener4", false, listener4.isReplyPubSubDomain()); + } + @Test public void testErrorHandlers() { ErrorHandler expected = this.context.getBean("testErrorHandler", ErrorHandler.class); diff --git a/spring-jms/src/test/java/org/springframework/jms/config/MessageListenerTestContainer.java b/spring-jms/src/test/java/org/springframework/jms/config/MessageListenerTestContainer.java index 650d9a7bec..0a716a0640 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/MessageListenerTestContainer.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/MessageListenerTestContainer.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. @@ -108,6 +108,11 @@ public class MessageListenerTestContainer return true; } + @Override + public boolean isReplyPubSubDomain() { + return isPubSubDomain(); + } + @Override public void afterPropertiesSet() { initializationInvoked = true; diff --git a/spring-jms/src/test/java/org/springframework/jms/config/MethodJmsListenerEndpointTests.java b/spring-jms/src/test/java/org/springframework/jms/config/MethodJmsListenerEndpointTests.java index 103c54222d..6b765d5299 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/MethodJmsListenerEndpointTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/MethodJmsListenerEndpointTests.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. @@ -241,6 +241,18 @@ public class MethodJmsListenerEndpointTests { assertListenerMethodInvocation(sample, methodName); } + @Test + public void processFromTopicAndReplyWithSendToQueue() throws JMSException { + String methodName = "processAndReplyWithSendTo"; + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setPubSubDomain(true); + container.setReplyPubSubDomain(false); + MessagingMessageListenerAdapter listener = createInstance(this.factory, + getListenerMethod(methodName, String.class), container); + processAndReplyWithSendTo(listener, false); + assertListenerMethodInvocation(sample, methodName); + } + @Test public void processAndReplyWithSendToTopic() throws JMSException { String methodName = "processAndReplyWithSendTo"; @@ -252,6 +264,17 @@ public class MethodJmsListenerEndpointTests { assertListenerMethodInvocation(sample, methodName); } + @Test + public void processFromQueueAndReplyWithSendToTopic() throws JMSException { + String methodName = "processAndReplyWithSendTo"; + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setReplyPubSubDomain(true); + MessagingMessageListenerAdapter listener = createInstance(this.factory, + getListenerMethod(methodName, String.class), container); + processAndReplyWithSendTo(listener, true); + assertListenerMethodInvocation(sample, methodName); + } + private void processAndReplyWithSendTo(MessagingMessageListenerAdapter listener, boolean pubSubDomain) throws JMSException { String body = "echo text"; String correlationId = "link-1234"; diff --git a/spring-jms/src/test/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManagerTests.java b/spring-jms/src/test/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManagerTests.java index e90fa5b468..c1d88177f5 100644 --- a/spring-jms/src/test/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManagerTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/listener/endpoint/JmsMessageEndpointManagerTests.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. @@ -37,6 +37,7 @@ public class JmsMessageEndpointManagerTests { config.setPubSubDomain(false); endpoint.setActivationSpecConfig(config); assertEquals(false, endpoint.isPubSubDomain()); + assertEquals(false, endpoint.isReplyPubSubDomain()); } @Test @@ -46,6 +47,18 @@ public class JmsMessageEndpointManagerTests { config.setPubSubDomain(true); endpoint.setActivationSpecConfig(config); assertEquals(true, endpoint.isPubSubDomain()); + assertEquals(true, endpoint.isReplyPubSubDomain()); + } + + @Test + public void pubSubDomainCustomForReply() { + JmsMessageEndpointManager endpoint = new JmsMessageEndpointManager(); + JmsActivationSpecConfig config = new JmsActivationSpecConfig(); + config.setPubSubDomain(true); + config.setReplyPubSubDomain(false); + endpoint.setActivationSpecConfig(config); + assertEquals(true, endpoint.isPubSubDomain()); + assertEquals(false, endpoint.isReplyPubSubDomain()); } @Test @@ -56,6 +69,14 @@ public class JmsMessageEndpointManagerTests { endpoint.isPubSubDomain(); } + @Test + public void isReplyPubSubDomainWithNoConfig() { + JmsMessageEndpointManager endpoint = new JmsMessageEndpointManager(); + + thrown.expect(IllegalStateException.class); // far from ideal + endpoint.isReplyPubSubDomain(); + } + @Test public void getMessageConverterNoConfig() { JmsMessageEndpointManager endpoint = new JmsMessageEndpointManager(); diff --git a/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml b/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml index 1e17a06329..cefdd9eea1 100644 --- a/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml +++ b/spring-jms/src/test/resources/org/springframework/jms/config/jmsNamespaceHandlerTests.xml @@ -3,12 +3,13 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd - http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd"> + http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.2.xsd"> + message-converter="testMessageConverter" destination-type="topic" response-destination-type="queue" + concurrency="5" prefetch="50" phase="77"> diff --git a/src/asciidoc/integration.adoc b/src/asciidoc/integration.adoc index cc87dc13fe..d89ea0a30c 100644 --- a/src/asciidoc/integration.adoc +++ b/src/asciidoc/integration.adoc @@ -2869,7 +2869,7 @@ describes all available attributes: | The name of the default response destination to send response messages to. This will be applied in case of a request message that does not carry a "JMSReplyTo" field. The type of this destination will be determined by the listener-container's - "destination-type" attribute. Note: This only applies to a listener method with a + "response-destination-type" attribute. Note: This only applies to a listener method with a return value, for which each result object will be converted into a response message. | subscription @@ -2958,6 +2958,10 @@ choices and message redelivery scenarios. and `subscriptionShared` properties of the container. The default is `queue` (i.e. disabling those 3 properties). +| response-destination-type +| The JMS destination type for responses: "queue", "topic". Default is the value of the + "destination-type" attribute. + | client-id | The JMS client id for this listener container. Needs to be specified when using durable subscriptions. @@ -3064,6 +3068,10 @@ table: and `subscriptionShared` properties of the container. The default is `queue` (i.e. disabling those 3 properties). +| response-destination-type +| The JMS destination type for responses: "queue", "topic". Default is the value of the + "destination-type" attribute. + | client-id | The JMS client id for this listener container. Needs to be specified when using durable subscriptions.