From c9d865176af262fbac3e2fb2bd00b758c6cb91d3 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 25 Jan 2010 23:44:57 +0000 Subject: [PATCH] INT-962 Exposing configuration properties for the JmsDestinationBackedMessageChannel's underlying MessageListener container and JmsTemplate. Many are now available in the namespace support, but some will still likely be added (the class itself now exposes *everything* that can be configured on both the MLC and template). --- .../JmsDestinationBackedMessageChannel.java | 164 ++++---- ...ListenerContainerConfigurationSupport.java | 367 ++++++++++++++++++ .../jms/config/JmsChannelParser.java | 129 +++++- .../jms/config/spring-integration-jms-2.0.xsd | 224 ++++++++++- 4 files changed, 809 insertions(+), 75 deletions(-) create mode 100644 org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java 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. - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +