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).

2010-01-25 23:44:57 +00:00
parent ece8d6c806
commit c9d865176a
4 changed files with 809 additions and 75 deletions
--- 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<Object> 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);
-	}
-
 }
--- 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
@@ -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<? extends AbstractMessageListenerContainer> 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<? extends AbstractMessageListenerContainer> 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();
+		}
+	}
+
+}
--- 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;
+		}
 	}

 }
--- 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 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="durable" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Boolean value indicating whether the Topic subscription is durable.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="client-id" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+	The JMS client id. Should be specified when using durable subscriptions.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="subscription" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+	The name for the durable subscription, if any.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>

 	<xsd:complexType name="channelType">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+	Base type for JMS Destination backed Message Channels (either 'channel' for a
+	Queue-backed channel or 'publish-subscribe-channel' for a Topic-backed channel).
+			]]></xsd:documentation>
+			<xsd:appinfo>
+				<tool:annotation>
+					<tool:exports type="org.springframework.integration.jms.JmsDestinationBackedMessageChannel"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
 		<xsd:attribute name="id" type="xsd:ID" use="required">
 			<xsd:annotation>
 				<xsd:documentation>
@@ -91,14 +123,198 @@
 	Reference to a JMS ConnectionFactory. If none is provided, the default
 	bean name for the reference will be "connectionFactory".
 				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="javax.jms.ConnectionFactory"/>
+					</tool:annotation>
+				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attribute name="destination-resolver" type="xsd:string">
 			<xsd:annotation>
-				<xsd:documentation>
-	Reference to a DestinationResolver. If none is provided, the default will
-	be a DynamicDestinationResolver.
-				</xsd:documentation>
+				<xsd:documentation><![CDATA[
+	A reference to the DestinationResolver strategy for resolving destination names.
+	Default is a DynamicDestinationResolver, using the JMS provider's queue/topic
+	name resolution. Alternatively, specify a reference to a JndiDestinationResolver
+	(typically in a J2EE environment).
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.jms.support.destination.DestinationResolver"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="container-type" default="default">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The type of this listener container: "default" or "simple", choosing
+	between DefaultMessageListenerContainer and SimpleMessageListenerContainer.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:NMTOKEN">
+					<xsd:enumeration value="default"/>
+					<xsd:enumeration value="simple"/>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="container-class" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	A custom listener container implementation class as fully qualified class name.
+	Default is Spring's standard DefaultMessageListenerContainer or
+	SimpleMessageListenerContainer, according to the "container-type" attribute.
+	Note that a custom container class will typically be a subclass of either of
+	those two Spring-provided standard container classes: Make sure that the
+	"container-type" attribute matches the actual base type that the custom class
+	derives from ("default" will usually be fine anyway, since most custom classes
+	will derive from DefaultMessageListenerContainer).
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation>
+						<tool:expected-type type="java.lang.Class"/>
+						<tool:assignable-to type="org.springframework.jms.listener.AbstractMessageListenerContainer"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="task-executor" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
+	JMS listener invokers. Default is a SimpleAsyncTaskExecutor in case of a
+	DefaultMessageListenerContainer, using internally managed threads. For a
+	SimpleMessageListenerContainer, listeners will always get invoked within the
+	JMS provider's receive thread by default.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="java.util.concurrent.Executor"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-converter" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	A reference to the MessageConverter strategy for converting between JMS Messages
+	and the Spring Integration Message payloads. Default is a SimpleMessageConverter.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.jms.support.converter.MessageConverter"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="error-handler" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	A reference to an ErrorHandler strategy for handling any uncaught Exceptions
+	that may occur during the execution of the underlying MessageListener.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.util.ErrorHandler"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="selector" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The JMS message selector for this channel's underlying MessageListener.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="cache" default="auto">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The cache level for JMS resources: "none", "connection", "session", "consumer"
+	or "auto". By default ("auto"), the cache level will effectively be "consumer",
+	unless an external transaction manager has been specified - in which case the
+	effective default will be "none" (assuming J2EE-style transaction management
+	where the given ConnectionFactory is an XA-aware pool).
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:NMTOKEN">
+					<xsd:enumeration value="none"/>
+					<xsd:enumeration value="connection"/>
+					<xsd:enumeration value="session"/>
+					<xsd:enumeration value="consumer"/>
+					<xsd:enumeration value="auto"/>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="acknowledge" default="transacted">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
+	A value of "transacted" effectively activates a locally transacted Session;
+	alternatively, specify an external "transaction-manager" via the corresponding
+	attribute. Default is "transacted".
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:NMTOKEN">
+					<xsd:enumeration value="auto"/>
+					<xsd:enumeration value="client"/>
+					<xsd:enumeration value="dups-ok"/>
+					<xsd:enumeration value="transacted"/>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="transaction-manager" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	A reference to an external PlatformTransactionManager (typically an
+	XA-based transaction coordinator, e.g. Spring's JtaTransactionManager).
+	If not specified, native acknowledging will be used (see "acknowledge" attribute).
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.transaction.PlatformTransactionManager"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="concurrency" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The number of concurrent sessions/consumers to start for each listener.
+	Can either be a simple number indicating the maximum number (e.g. "5")
+	or a range indicating the lower as well as the upper limit (e.g. "3-5").
+	Note that a specified minimum is just a hint and might be ignored at runtime.
+	Default is 1; keep concurrency limited to 1 in case of a topic listener
+	or if message ordering is important; consider raising it for general queues.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="prefetch" type="xsd:int">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The maximum number of messages to load into a single session.
+	Note that raising this number might lead to starvation of concurrent consumers!
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="auto-startup" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	Boolean value indicating whether this channel's listener container should start automatically.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="phase" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The lifecycle phase within which this channel's listener container should start and stop.
+	The lower the value the earlier this container will start and the later it will stop. The
+	default is Integer.MAX_VALUE meaning the container will start as late as possible
+	and stop as soon as possible.
+				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
 	</xsd:complexType>