diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/AbstractJmsTemplateBasedAdapter.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/AbstractJmsTemplateBasedAdapter.java index bcb75a57d4..2a1fa17ac7 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/AbstractJmsTemplateBasedAdapter.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/AbstractJmsTemplateBasedAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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,8 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea private volatile String destinationName; + private volatile boolean pubSubDomain; + private volatile DestinationResolver destinationResolver; private volatile JmsTemplate jmsTemplate; @@ -82,6 +84,10 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea this.destinationName = destinationName; } + public void setPubSubDomain(boolean pubSubDomain) { + this.pubSubDomain = pubSubDomain; + } + public void setDestinationResolver(DestinationResolver destinationResolver) { this.destinationResolver = destinationResolver; } @@ -126,6 +132,7 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea } else { jmsTemplate.setDefaultDestinationName(this.destinationName); + jmsTemplate.setPubSubDomain(this.pubSubDomain); } if (this.destinationResolver != null) { jmsTemplate.setDestinationResolver(this.destinationResolver); diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 8478e09f0a..d1c9ccc1c4 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -16,6 +16,7 @@ package org.springframework.integration.jms; +import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.InvalidDestinationException; import javax.jms.JMSException; @@ -55,6 +56,12 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL private volatile Object defaultReplyDestination; + private volatile long replyTimeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE; + + private volatile int replyPriority = javax.jms.Message.DEFAULT_PRIORITY; + + private volatile int replyDeliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE; + private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver(); private volatile JmsHeaderMapper headerMapper; @@ -127,6 +134,30 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL this.defaultReplyDestination = new DestinationNameHolder(destinationName, true); } + /** + * Specify the time-to-live property for JMS reply Messages. + * @see javax.jms.MessageProducer#setTimeToLive(long) + */ + public void setReplyTimeToLive(long replyTimeToLive) { + this.replyTimeToLive = replyTimeToLive; + } + + /** + * Specify the priority value for JMS reply Messages. + * @see javax.jms.MessageProducer#setPriority(int) + */ + public void setReplyPriority(int replyPriority) { + this.replyPriority = replyPriority; + } + + /** + * Specify the delivery mode for JMS reply Messages. + * @see javax.jms.MessageProducer#setDeliveryMode(int) + */ + public void setReplyDeliveryPersistent(boolean replyDeliveryPersistent) { + this.replyDeliveryMode = replyDeliveryPersistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT; + } + /** * Set the DestinationResolver that should be used to resolve reply * destination names for this listener. @@ -214,6 +245,9 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID()); } MessageProducer producer = session.createProducer(destination); + producer.setTimeToLive(this.replyTimeToLive); + producer.setPriority(this.replyPriority); + producer.setDeliveryMode(this.replyDeliveryMode); try { producer.send(jmsReply); } diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java index c3a97cc830..c0a8d3d365 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. @@ -42,6 +42,10 @@ abstract class JmsAdapterParserUtils { static final String DESTINATION_NAME_ATTRIBUTE = "destination-name"; + static final String PUB_SUB_DOMAIN_ATTRIBUTE = "pub-sub-domain"; + + static final String PUB_SUB_DOMAIN_PROPERTY = "pubSubDomain"; + static final String DESTINATION_NAME_PROPERTY = "destinationName"; static final String HEADER_MAPPER_ATTRIBUTE = "header-mapper"; diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java index 6c4e320873..a3e2c59d4e 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java @@ -45,36 +45,48 @@ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChanne @Override protected String parseSource(Element element, ParserContext parserContext) { + Object source = parserContext.extractSource(element); BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( "org.springframework.integration.jms.JmsDestinationPollingSource"); String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE); String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); + String pubSubDomain = element.getAttribute(JmsAdapterParserUtils.PUB_SUB_DOMAIN_ATTRIBUTE); String headerMapper = element.getAttribute(JmsAdapterParserUtils.HEADER_MAPPER_ATTRIBUTE); + boolean hasDestinationRef = StringUtils.hasText(destination); + boolean hasDestinationName = StringUtils.hasText(destinationName); if (StringUtils.hasText(jmsTemplate)) { if (element.hasAttribute(JmsAdapterParserUtils.CONNECTION_FACTORY_ATTRIBUTE) || - element.hasAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE) || - element.hasAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE)) { - throw new BeanCreationException( + hasDestinationRef || hasDestinationName) { + parserContext.getReaderContext().error( "When providing '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + "', none of '" + JmsAdapterParserUtils.CONNECTION_FACTORY_ATTRIBUTE + "', '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "', or '" + - JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' should be provided."); + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' are allowed.", + source); } builder.addConstructorArgReference(jmsTemplate); } - else if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { + else if (hasDestinationRef || hasDestinationName) { builder.addConstructorArgReference(JmsAdapterParserUtils.determineConnectionFactoryBeanName(element, parserContext)); - if (StringUtils.hasText(destination)) { + if (hasDestinationRef) { + if (hasDestinationName) { + parserContext.getReaderContext().error("The 'destination-name' " + + "and 'destination' attributes are mutually exclusive.", source); + } builder.addConstructorArgReference(destination); } - else if (StringUtils.hasText(destinationName)) { + else if (hasDestinationName) { builder.addConstructorArgValue(destinationName); + if (StringUtils.hasText(pubSubDomain)) { + builder.addPropertyValue(JmsAdapterParserUtils.PUB_SUB_DOMAIN_PROPERTY, pubSubDomain); + } } } else { - throw new BeanCreationException("either a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + "' or one of '" + - JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + + throw new BeanCreationException("either a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + + "' or one of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' attributes must be provided for a polling JMS adapter"); } if (StringUtils.hasText(headerMapper)) { diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index c4222e60ba..f73d59358e 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -41,12 +41,18 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition private static final String DEFAULT_REPLY_TOPIC_NAME_ATTRIB = "default-reply-topic-name"; + private static final String REPLY_TIME_TO_LIVE = "reply-time-to-live"; + + private static final String REPLY_PRIORITY = "reply-priority"; + + private static final String REPLY_DELIVERY_PERSISTENT = "reply-delivery-persistent"; + private static String[] containerAttributes = new String[] { JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY, JmsAdapterParserUtils.DESTINATION_ATTRIBUTE, JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE, - "destination-resolver", "transaction-manager", "pub-sub-domain", + "destination-resolver", "transaction-manager", "concurrent-consumers", "max-concurrent-consumers", "max-messages-per-task", "idle-task-execution-limit", "selector" }; @@ -101,17 +107,22 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition String destinationNameAttribute = this.expectReply ? "request-destination-name" : "destination-name"; String destination = element.getAttribute(destinationAttribute); String destinationName = element.getAttribute(destinationNameAttribute); - if (!(StringUtils.hasText(destination) ^ StringUtils.hasText(destinationName))) { + boolean hasDestination = StringUtils.hasText(destination); + boolean hasDestinationName = StringUtils.hasText(destinationName); + if (!(hasDestination ^ hasDestinationName)) { parserContext.getReaderContext().error( - "Exactly one of '" + destinationAttribute + "' or '" + destinationNameAttribute + "' is required.", element); + "Exactly one of '" + destinationAttribute + + "' or '" + destinationNameAttribute + "' is required.", element); } builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY, JmsAdapterParserUtils.determineConnectionFactoryBeanName(element, parserContext)); - if (StringUtils.hasText(destination)) { + if (hasDestination) { builder.addPropertyReference("destination", destination); } else { builder.addPropertyValue("destinationName", destinationName); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + "request-pub-sub-domain", "pubSubDomain"); } Integer acknowledgeMode = JmsAdapterParserUtils.parseAcknowledgeMode(element, parserContext); if (acknowledgeMode != null) { @@ -124,7 +135,6 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "pub-sub-domain"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-concurrent-consumers"); @@ -163,6 +173,9 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, DEFAULT_REPLY_QUEUE_NAME_ATTRIB); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, DEFAULT_REPLY_TOPIC_NAME_ATTRIB); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REPLY_TIME_TO_LIVE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REPLY_PRIORITY); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REPLY_DELIVERY_PERSISTENT); } else { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "requestChannel"); diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java index 8eaa110000..fd7d33eb25 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. @@ -27,7 +27,7 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.util.StringUtils; /** - * Parser for the <jms-target/> element. + * Parser for the <outbound-channel-adapter/> element of the jms namespace. * * @author Mark Fisher */ @@ -41,16 +41,17 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); String headerMapper = element.getAttribute(JmsAdapterParserUtils.HEADER_MAPPER_ATTRIBUTE); + boolean hasDestinationRef = StringUtils.hasText(destination); + boolean hasDestinationName = StringUtils.hasText(destinationName); if (StringUtils.hasText(jmsTemplate)) { if (element.hasAttribute(JmsAdapterParserUtils.CONNECTION_FACTORY_ATTRIBUTE) || - element.hasAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE) || - element.hasAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE)) { + hasDestinationRef || hasDestinationName) { throw new BeanCreationException("When providing a 'jms-template' reference, none of " + "'connection-factory', 'destination', or 'destination-name' should be provided."); } builder.addPropertyReference(JmsAdapterParserUtils.JMS_TEMPLATE_PROPERTY, jmsTemplate); } - else if (StringUtils.hasText(destination) ^ StringUtils.hasText(destinationName)) { + else if (hasDestinationRef ^ hasDestinationName) { builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY, JmsAdapterParserUtils.determineConnectionFactoryBeanName(element, parserContext)); if (StringUtils.hasText(destination)) { @@ -58,6 +59,8 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap } else { builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName); + IntegrationNamespaceUtils.setValueIfAttributeDefined( + builder, element, JmsAdapterParserUtils.PUB_SUB_DOMAIN_ATTRIBUTE); } } else { diff --git a/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd b/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd index ceca4992b7..1406ff5eae 100644 --- a/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd +++ b/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd @@ -330,6 +330,11 @@ + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use 'destination-name' and 'pub-sub-domain' which will rely upon the + DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -338,6 +343,7 @@ + @@ -390,6 +396,11 @@ + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use 'destination-name' and 'pub-sub-domain' which will rely upon the + DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -398,6 +409,7 @@ + @@ -442,6 +454,11 @@ + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use 'request-destination-name' and 'request-pub-sub-domain' which will rely + upon the DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -450,8 +467,14 @@ + + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use either 'default-reply-queue-name' or 'default-reply-topic-name' which + will rely upon the DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -496,6 +519,9 @@ + + + @@ -534,6 +560,11 @@ + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use 'request-destination-name' and 'request-pub-sub-domain' which will rely + upon the DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -542,8 +573,14 @@ + + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use 'reply-destination-name' and 'reply-pub-sub-domain' which will rely + upon the DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -552,6 +589,7 @@ + @@ -561,7 +599,6 @@ - @@ -621,6 +658,11 @@ + + A reference to a javax.jms.Destination by bean name. As an alternative to a bean + reference, use 'destination-name' and 'pub-sub-domain' which will rely upon the + DestinationResolver strategy (DynamicDestinationResolver by default). + @@ -629,6 +671,7 @@ + @@ -807,7 +850,6 @@ - diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java index b1d54eddf3..b7ba2240a7 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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,8 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import javax.jms.DeliveryMode; + import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; @@ -262,4 +264,16 @@ public class JmsInboundGatewayParserTests { gateway.stop(); } + @Test + public void testGatewayWithReplyQosProperties() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "inboundGatewayWithReplyQos.xml", this.getClass()); + JmsMessageDrivenEndpoint gateway = context.getBean("gatewayWithReplyQos", JmsMessageDrivenEndpoint.class); + DirectFieldAccessor accessor = new DirectFieldAccessor( + new DirectFieldAccessor(gateway).getPropertyValue("listener")); + assertEquals(12345L, accessor.getPropertyValue("replyTimeToLive")); + assertEquals(7, accessor.getPropertyValue("replyPriority")); + assertEquals(DeliveryMode.NON_PERSISTENT, accessor.getPropertyValue("replyDeliveryMode")); + } + } diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithReplyQos.xml b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithReplyQos.xml new file mode 100644 index 0000000000..915a728b2b --- /dev/null +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithReplyQos.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsGatewayWithContainerSettings.xml b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsGatewayWithContainerSettings.xml index 9e9d43c559..cc8e17c82a 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsGatewayWithContainerSettings.xml +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsGatewayWithContainerSettings.xml @@ -17,25 +17,29 @@