diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java index 92701fba18..17c07b9f34 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.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. @@ -19,7 +19,6 @@ package org.springframework.integration.jms; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.MessageListener; -import javax.jms.Topic; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; @@ -36,10 +35,9 @@ import org.springframework.integration.gateway.SimpleMessageMapper; import org.springframework.integration.message.InboundMessageMapper; import org.springframework.integration.message.MessageHandler; import org.springframework.jms.core.JmsTemplate; -import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.support.converter.MessageConverter; import org.springframework.jms.support.destination.DestinationResolver; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.util.Assert; /** * A {@link MessageChannel} implementation that is actually backed by a JMS @@ -53,13 +51,11 @@ import org.springframework.util.Assert; * @author Mark Fisher * @since 2.0 */ -public class JmsDestinationBackedMessageChannel implements SubscribableChannel, MessageListener, - BeanNameAware, SmartLifecycle, InitializingBean { +public class JmsDestinationBackedMessageChannel extends MessageListenerContainerConfigurationSupport + implements SubscribableChannel, MessageListener, BeanNameAware, SmartLifecycle, InitializingBean { private final JmsTemplate jmsTemplate = new JmsTemplate(); - private final DefaultMessageListenerContainer container = new DefaultMessageListenerContainer(); - private final InboundMessageMapper mapper = new SimpleMessageMapper(); private volatile MessageDispatcher dispatcher; @@ -68,35 +64,109 @@ public class JmsDestinationBackedMessageChannel implements SubscribableChannel, public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, Destination destination) { - Assert.notNull(connectionFactory, "connectionFactory must not be null"); - Assert.notNull(destination, "destination must not be null"); - this.jmsTemplate.setConnectionFactory(connectionFactory); - this.jmsTemplate.setDefaultDestination(destination); - this.initDispatcher(destination instanceof Topic); + this.setConnectionFactory(connectionFactory); + this.setDestination(destination); } public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, String destinationName, boolean isPubSub) { - this(connectionFactory, destinationName, isPubSub, null); + this.setConnectionFactory(connectionFactory); + this.setDestinationName(destinationName); + this.setPubSubDomain(isPubSub); } - public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, String destinationName, boolean isPubSub, DestinationResolver destinationResolver) { - Assert.notNull(connectionFactory, "connectionFactory must not be null"); - Assert.hasText(destinationName, "destinationName is required"); + + @Override + public void setConnectionFactory(ConnectionFactory connectionFactory) { + super.setConnectionFactory(connectionFactory); this.jmsTemplate.setConnectionFactory(connectionFactory); - if (destinationResolver != null) { - this.jmsTemplate.setDestinationResolver(destinationResolver); - } - this.jmsTemplate.setDefaultDestinationName(destinationName); - this.jmsTemplate.setPubSubDomain(isPubSub); - this.initDispatcher(isPubSub); } + @Override + public void setDestination(Destination destination) { + super.setDestination(destination); + this.jmsTemplate.setDefaultDestination(destination); + } + + @Override + public void setDestinationName(String destinationName) { + super.setDestinationName(destinationName); + this.jmsTemplate.setDefaultDestinationName(destinationName); + } + + @Override + public void setDestinationResolver(DestinationResolver destinationResolver) { + super.setDestinationResolver(destinationResolver); + this.jmsTemplate.setDestinationResolver(destinationResolver); + } + + @Override + public void setPubSubDomain(boolean pubSubDomain) { + super.setPubSubDomain(pubSubDomain); + this.jmsTemplate.setPubSubDomain(pubSubDomain); + } + + public void setDeliveryPersistent(boolean deliveryPersistent) { + this.jmsTemplate.setDeliveryPersistent(deliveryPersistent); + } + + public void setExplicitQosEnabled(boolean explicitQosEnabled) { + this.jmsTemplate.setExplicitQosEnabled(explicitQosEnabled); + } + + public void setMessageConverter(MessageConverter messageConverter) { + this.jmsTemplate.setMessageConverter(messageConverter); + } + + public void setMessageIdEnabled(boolean messageIdEnabled) { + this.jmsTemplate.setMessageIdEnabled(messageIdEnabled); + } + + public void setMessageTimestampEnabled(boolean messageTimestampEnabled) { + this.jmsTemplate.setMessageTimestampEnabled(messageTimestampEnabled); + } + + public void setPriority(int priority) { + this.jmsTemplate.setPriority(priority); + } + + @Override + public void setPubSubNoLocal(boolean pubSubNoLocal) { + super.setPubSubNoLocal(pubSubNoLocal); + this.jmsTemplate.setPubSubNoLocal(pubSubNoLocal); + } + + @Override + public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) { + super.setSessionAcknowledgeMode(sessionAcknowledgeMode); + this.jmsTemplate.setSessionAcknowledgeMode(sessionAcknowledgeMode); + } + + @Override + public void setSessionTransacted(boolean sessionTransacted) { + super.setSessionTransacted(sessionTransacted); + this.jmsTemplate.setSessionTransacted(sessionTransacted); + } + + public void setTimeToLive(long timeToLive) { + this.jmsTemplate.setTimeToLive(timeToLive); + } public void setBeanName(String beanName) { this.name = beanName; } - private void initDispatcher(boolean isPubSub) { + @Override + public void afterPropertiesSet() throws Exception { + super.afterPropertiesSet(); + AbstractMessageListenerContainer container = this.getListenerContainer(); + this.configureDispatcher(container.isPubSubDomain()); + container.setMessageListener(this); + if (!container.isActive()) { + container.afterPropertiesSet(); + } + } + + private void configureDispatcher(boolean isPubSub) { if (isPubSub) { this.dispatcher = new BroadcastingDispatcher(); } @@ -107,24 +177,6 @@ public class JmsDestinationBackedMessageChannel implements SubscribableChannel, } } - public void setTransactionManager(PlatformTransactionManager transactionManager) { - this.container.setTransactionManager(transactionManager); - } - - public void afterPropertiesSet() throws Exception { - this.container.setConnectionFactory(this.jmsTemplate.getConnectionFactory()); - Destination destination = this.jmsTemplate.getDefaultDestination(); - if (destination != null) { - this.container.setDestination(destination); - } - else { - this.container.setDestinationName(this.jmsTemplate.getDefaultDestinationName()); - this.container.setPubSubDomain(this.jmsTemplate.isPubSubDomain()); - } - this.container.setMessageListener(this); - this.container.afterPropertiesSet(); - } - public String getName() { return this.name; } @@ -159,30 +211,4 @@ public class JmsDestinationBackedMessageChannel implements SubscribableChannel, } } - // SmartLifecycle implementation (delegates to the MessageListener container) - - public int getPhase() { - return this.container.getPhase(); - } - - public boolean isAutoStartup() { - return this.container.isAutoStartup(); - } - - public boolean isRunning() { - return this.container.isRunning(); - } - - public void start() { - this.container.start(); - } - - public void stop() { - this.container.stop(); - } - - public void stop(Runnable callback) { - this.container.stop(callback); - } - } diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java new file mode 100644 index 0000000000..89d60fb612 --- /dev/null +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java @@ -0,0 +1,367 @@ +/* + * 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.jms; + +import java.util.concurrent.Executor; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.ExceptionListener; +import javax.jms.Session; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.SimpleMessageListenerContainer; +import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.ErrorHandler; + +/** + * A base class for managing configurable properties of a MessageListenerContainer. + * + * @author Mark Fisher + * @since 2.0 + */ +abstract class MessageListenerContainerConfigurationSupport implements InitializingBean { + + private volatile AbstractMessageListenerContainer container; + + private volatile Class containerType; + + private volatile boolean acceptMessagesWhileStopping; + + private volatile boolean autoStartup = true; + + private volatile String cacheLevelName; + + private volatile String clientId; + + private volatile Integer concurrentConsumers; + + private volatile ConnectionFactory connectionFactory; + + private volatile Destination destination; + + private volatile String destinationName; + + private volatile DestinationResolver destinationResolver; + + private volatile String durableSubscriptionName; + + private volatile ErrorHandler errorHandler; + + private volatile ExceptionListener exceptionListener; + + private volatile Boolean exposeListenerSession; + + private volatile Integer idleTaskExecutionLimit; + + private volatile Integer maxConcurrentConsumers; + + private volatile Integer maxMessagesPerTask; + + private volatile String messageSelector; + + private volatile Integer phase; + + private volatile Boolean pubSubDomain; + + private volatile boolean pubSubNoLocal; + + private volatile Long receiveTimeout; + + private volatile Long recoveryInterval; + + /** + * This value differs from the container implementations' default (which is AUTO_ACKNOWLEDGE) + */ + private volatile int sessionAcknowledgeMode = Session.SESSION_TRANSACTED; + + /** + * This value differs from the container implementations' default (which is false). + */ + private volatile boolean sessionTransacted = true; + + private volatile boolean subscriptionDurable; + + private volatile Executor taskExecutor; + + private volatile PlatformTransactionManager transactionManager; + + private volatile String transactionName; + + private volatile Integer transactionTimeout; + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + public void setAcceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { + this.acceptMessagesWhileStopping = acceptMessagesWhileStopping; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setCacheLevelName(String cacheLevelName) { + this.cacheLevelName = cacheLevelName; + } + + public void setClientId(String clientId) { + this.clientId = clientId; + } + + public void setConcurrentConsumers(int concurrentConsumers) { + this.concurrentConsumers = concurrentConsumers; + } + + public void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + } + + public void setContainerType(Class containerType) { + this.containerType = containerType; + } + + public void setDestination(Destination destination) { + this.destination = destination; + } + + public void setDestinationName(String destinationName) { + this.destinationName = destinationName; + } + + public void setDestinationResolver(DestinationResolver destinationResolver) { + this.destinationResolver = destinationResolver; + } + + public void setDurableSubscriptionName(String durableSubscriptionName) { + this.durableSubscriptionName = durableSubscriptionName; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public void setExceptionListener(ExceptionListener exceptionListener) { + this.exceptionListener = exceptionListener; + } + + public void setExposeListenerSession(boolean exposeListenerSession) { + this.exposeListenerSession = exposeListenerSession; + } + + public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) { + this.idleTaskExecutionLimit = idleTaskExecutionLimit; + } + + public void setMaxConcurrentConsumers(int maxConcurrentConsumers) { + this.maxConcurrentConsumers = maxConcurrentConsumers; + } + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public void setMessageSelector(String messageSelector) { + this.messageSelector = messageSelector; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + public void setPubSubDomain(boolean pubSubDomain) { + this.pubSubDomain = pubSubDomain; + } + + public void setPubSubNoLocal(boolean pubSubNoLocal) { + this.pubSubNoLocal = pubSubNoLocal; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public void setRecoveryInterval(long recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + + public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) { + this.sessionAcknowledgeMode = sessionAcknowledgeMode; + } + + public void setSessionTransacted(boolean sessionTransacted) { + this.sessionTransacted = sessionTransacted; + } + + public void setSubscriptionDurable(boolean subscriptionDurable) { + this.subscriptionDurable = subscriptionDurable; + } + + public void setTaskExecutor(Executor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setTransactionName(String transactionName) { + this.transactionName = transactionName; + } + + public void setTransactionTimeout(int transactionTimeout) { + this.transactionTimeout = transactionTimeout; + } + + AbstractMessageListenerContainer getListenerContainer() { + if (!this.initialized) { + try { + this.initialize(); + } + catch (Exception e) { + throw new IllegalStateException("failed to initialize listener container", e); + } + } + return this.container; + } + + public void afterPropertiesSet() throws Exception { + this.initialize(); + } + + private void initialize() throws Exception { + synchronized (this.initializationMonitor) { + if (this.initialized) { + return; + } + if (this.containerType == null) { + this.containerType = DefaultMessageListenerContainer.class; + } + this.container = this.containerType.newInstance(); + this.container.setAcceptMessagesWhileStopping(this.acceptMessagesWhileStopping); + this.container.setAutoStartup(this.autoStartup); + this.container.setClientId(this.clientId); + this.container.setConnectionFactory(this.connectionFactory); + if (this.destination != null) { + this.container.setDestination(this.destination); + } + if (this.destinationName != null) { + this.container.setDestinationName(this.destinationName); + } + if (this.destinationResolver != null) { + this.container.setDestinationResolver(this.destinationResolver); + } + this.container.setDurableSubscriptionName(this.durableSubscriptionName); + this.container.setErrorHandler(this.errorHandler); + this.container.setExceptionListener(this.exceptionListener); + if (this.exposeListenerSession != null) { + this.container.setExposeListenerSession(this.exposeListenerSession); + } + this.container.setMessageSelector(this.messageSelector); + if (this.phase != null) { + this.container.setPhase(this.phase); + } + if (this.pubSubDomain != null) { + this.container.setPubSubDomain(this.pubSubDomain); + } + this.container.setSessionAcknowledgeMode(this.sessionAcknowledgeMode); + this.container.setSessionTransacted(this.sessionTransacted); + this.container.setSubscriptionDurable(this.subscriptionDurable); + if (this.container instanceof DefaultMessageListenerContainer) { + DefaultMessageListenerContainer dmlc = (DefaultMessageListenerContainer) this.container; + if (this.cacheLevelName != null) { + dmlc.setCacheLevelName(this.cacheLevelName); + } + if (this.concurrentConsumers != null) { + dmlc.setConcurrentConsumers(this.concurrentConsumers); + } + if (this.idleTaskExecutionLimit != null) { + dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit); + } + if (this.maxConcurrentConsumers != null) { + dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers); + } + if (this.maxMessagesPerTask != null) { + dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask); + } + dmlc.setPubSubNoLocal(this.pubSubNoLocal); + if (this.receiveTimeout != null) { + dmlc.setReceiveTimeout(this.receiveTimeout); + } + if (this.recoveryInterval != null) { + dmlc.setRecoveryInterval(this.recoveryInterval); + } + dmlc.setTaskExecutor(this.taskExecutor); + dmlc.setTransactionManager(this.transactionManager); + if (this.transactionName != null) { + dmlc.setTransactionName(this.transactionName); + } + if (this.transactionTimeout != null) { + dmlc.setTransactionTimeout(this.transactionTimeout); + } + } + else if (this.container instanceof SimpleMessageListenerContainer) { + SimpleMessageListenerContainer smlc = (SimpleMessageListenerContainer) this.container; + if (this.concurrentConsumers != null) { + smlc.setConcurrentConsumers(this.concurrentConsumers); + } + smlc.setPubSubNoLocal(this.pubSubNoLocal); + smlc.setTaskExecutor(this.taskExecutor); + } + this.initialized = true; + } + } + + // SmartLifecycle implementation (delegates to the MessageListener container) + + public int getPhase() { + return this.getListenerContainer().getPhase(); + } + + public boolean isAutoStartup() { + return this.autoStartup; + } + + public boolean isRunning() { + return this.initialized && this.container.isRunning(); + } + + public void start() { + this.getListenerContainer().start(); + } + + public void stop() { + if (this.isRunning()) { + this.container.stop(); + } + } + + public void stop(Runnable callback) { + if (this.isRunning()) { + this.container.stop(callback); + } + else { + callback.run(); + } + } + +} diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java index 387c9fe3b8..549d0ae6c2 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.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,11 +16,15 @@ package org.springframework.integration.jms.config; +import javax.jms.Session; + import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** @@ -32,6 +36,13 @@ import org.springframework.util.StringUtils; */ public class JmsChannelParser extends AbstractSingleBeanDefinitionParser { + private final static String CONTAINER_TYPE_ATTRIBUTE = "container-type"; + + private final static String CONTAINER_CLASS_ATTRIBUTE = "container-class"; + + private final static String ACKNOWLEDGE_ATTRIBUTE = "acknowledge"; + + @Override protected String getBeanClassName(Element element) { return "org.springframework.integration.jms.JmsDestinationBackedMessageChannel"; @@ -50,6 +61,67 @@ public class JmsChannelParser extends AbstractSingleBeanDefinitionParser { else if ("publish-subscribe-channel".equals(element.getLocalName())) { this.parseDestination(element, parserContext, builder, "topic"); } + String containerType = element.getAttribute(CONTAINER_TYPE_ATTRIBUTE); + String containerClass = element.getAttribute(CONTAINER_CLASS_ATTRIBUTE); + Assert.isTrue(!(StringUtils.hasText(containerType) && StringUtils.hasText(containerClass)), + "At most one of '" + CONTAINER_TYPE_ATTRIBUTE + "' or '" + CONTAINER_CLASS_ATTRIBUTE + "' may be configured."); + if ("default".equals(containerType)) { + containerClass = "org.springframework.jms.listener.DefaultMessageListenerContainer"; + } + else if ("simple".equals(containerType)) { + containerClass = "org.springframework.jms.listener.SimpleMessageListenerContainer"; + } + if (StringUtils.hasText(containerClass)) { + builder.addPropertyValue("containerType", containerClass); + if (containerClass.contains("DefaultMessageListenerContainer")) { + containerType = "default"; + } + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + String cache = element.getAttribute("cache"); + if (StringUtils.hasText(cache)) { + if (containerType.startsWith("simple")) { + if (!("auto".equals(cache) || "consumer".equals(cache))) { + parserContext.getReaderContext().warning( + "'cache' attribute not actively supported for listener container of type \"simple\". " + + "Effective runtime behavior will be equivalent to \"consumer\" / \"auto\".", element); + } + } + else { + builder.addPropertyValue("cacheLevelName", "CACHE_" + cache.toUpperCase()); + } + } + Integer acknowledgeMode = this.parseAcknowledgeMode(element, parserContext); + if (acknowledgeMode != null) { + if (acknowledgeMode == Session.SESSION_TRANSACTED) { + builder.addPropertyValue("sessionTransacted", Boolean.TRUE); + } + else { + builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); + } + } + int[] concurrency = parseConcurrency(element, parserContext); + if (concurrency != null) { + if (containerType.startsWith("default")) { + builder.addPropertyValue("concurrentConsumers", concurrency[0]); + builder.addPropertyValue("maxConcurrentConsumers", concurrency[1]); + } + else { + builder.addPropertyValue("concurrentConsumers", concurrency[1]); + } + } + String prefetch = element.getAttribute("prefetch"); + if (StringUtils.hasText(prefetch)) { + if (containerType.startsWith("default")) { + builder.addPropertyValue("maxMessagesPerTask", new Integer(prefetch)); + } + } } private void parseDestination(Element element, ParserContext parserContext, BeanDefinitionBuilder builder, String type) { @@ -70,9 +142,62 @@ public class JmsChannelParser extends AbstractSingleBeanDefinitionParser { builder.addConstructorArgValue(isPubSub); String destinationResolver = element.getAttribute("destination-resolver"); if (StringUtils.hasText(destinationResolver)) { - builder.addConstructorArgReference(destinationResolver); + builder.addPropertyReference("destinationResolver", destinationResolver); } } + if (isPubSub) { + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "durable", "subscriptionDurable"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "subscription", "durableSubscriptionName"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id"); + } + } + + private Integer parseAcknowledgeMode(Element ele, ParserContext parserContext) { + String acknowledge = ele.getAttribute(ACKNOWLEDGE_ATTRIBUTE); + if (StringUtils.hasText(acknowledge)) { + int acknowledgeMode = Session.AUTO_ACKNOWLEDGE; + if ("transacted".equals(acknowledge)) { + acknowledgeMode = Session.SESSION_TRANSACTED; + } + else if ("dups-ok".equals(acknowledge)) { + acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE; + } + else if ("client".equals(acknowledge)) { + acknowledgeMode = Session.CLIENT_ACKNOWLEDGE; + } + else if (!"auto".equals(acknowledge)) { + parserContext.getReaderContext().error("Invalid JMS Channel 'acknowledge' setting [" + + acknowledge + "]: only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.", ele); + } + return acknowledgeMode; + } + else { + return null; + } + } + + private int[] parseConcurrency(Element ele, ParserContext parserContext) { + String concurrency = ele.getAttribute("concurrency"); + if (!StringUtils.hasText(concurrency)) { + return null; + } + try { + int separatorIndex = concurrency.indexOf('-'); + if (separatorIndex != -1) { + int[] result = new int[2]; + result[0] = Integer.parseInt(concurrency.substring(0, separatorIndex)); + result[1] = Integer.parseInt(concurrency.substring(separatorIndex + 1, concurrency.length())); + return result; + } + else { + return new int[] {1, Integer.parseInt(concurrency)}; + } + } + catch (NumberFormatException ex) { + parserContext.getReaderContext().error("Invalid concurrency value [" + concurrency + "]: only " + + "single maximum integer (e.g. \"5\") and minimum-maximum combo (e.g. \"3-5\") supported.", ele, ex); + return null; + } } } 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 7456e97f93..ceca4992b7 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 @@ -72,12 +72,44 @@ + + + + Boolean value indicating whether the Topic subscription is durable. + + + + + + + + + + + + + + + + + + + + + @@ -91,14 +123,198 @@ Reference to a JMS ConnectionFactory. If none is provided, the default bean name for the reference will be "connectionFactory". + + + + + - - Reference to a DestinationResolver. If none is provided, the default will - be a DynamicDestinationResolver. - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +