INT-3567: JMS Shared Subscriptions

JIRA: https://jira.spring.io/browse/INT-3567

Namespace support for JMS 2.0 shared subscriptions.

INT-3567: AsciiDoc

* `What's New` polishing
* `TcpConnectionEventInboundChannelAdapterParser` deprecation warn fix

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionEventInboundChannelAdapterParser.java

@@ -16,14 +16,13 @@
 
 package org.springframework.integration.ip.config;
 
-import org.w3c.dom.Element;
-
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
-import org.springframework.integration.ip.tcp.connection.TcpConnectionEventListeningMessageProducer;
+
+import org.w3c.dom.Element;
 
 /**
  * @author Gary Russell
@@ -35,9 +34,10 @@ import org.springframework.integration.ip.tcp.connection.TcpConnectionEventListe
 public class TcpConnectionEventInboundChannelAdapterParser extends AbstractChannelAdapterParser {
 
 	@Override
+	@SuppressWarnings("deprecation")
 	protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
 		BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
-				.rootBeanDefinition(TcpConnectionEventListeningMessageProducer.class);
+				.rootBeanDefinition(org.springframework.integration.ip.tcp.connection.TcpConnectionEventListeningMessageProducer.class);
 		adapterBuilder.addPropertyReference("outputChannel", channelName);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(adapterBuilder, element, "error-channel", "errorChannel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "event-types");

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -117,6 +117,8 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 
 	private volatile String beanName;
 
+	private volatile boolean subscriptionShared;
+
 	/**
 	 * This value differs from the container implementations' default (which is AUTO_ACKNOWLEDGE)
 	 */
@@ -354,6 +356,10 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		this.maxSubscribers = maxSubscribers;
 	}
 
+	public void setSubscriptionShared(boolean subscriptionShared) {
+		this.subscriptionShared = subscriptionShared;
+	}
+
 	@Override
 	public void setBeanName(String name) {
 		this.beanName = name;
@@ -368,7 +374,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 	protected AbstractJmsChannel createInstance() throws Exception {
 		this.initializeJmsTemplate();
 		if (this.messageDriven) {
-			this.container = this.createContainer();
+			this.container = createContainer();
 			SubscribableJmsChannel subscribableJmsChannel = new SubscribableJmsChannel(this.container, this.jmsTemplate);
 			subscribableJmsChannel.setMaxSubscribers(this.maxSubscribers);
 			this.channel = subscribableJmsChannel;
@@ -438,6 +444,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		container.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
 		container.setSessionTransacted(this.sessionTransacted);
 		container.setSubscriptionDurable(this.subscriptionDurable);
+		container.setSubscriptionShared(this.subscriptionShared);
 
 
 

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java

@@ -156,6 +156,7 @@ public class JmsChannelParser extends AbstractChannelParser {
 		if (isPubSub) {
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "durable", "subscriptionDurable");
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "subscription", "durableSubscriptionName");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "subscription-shared");
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id");
 		}
 	}

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java

@@ -67,6 +67,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 		"receive-timeout", "recovery-interval",
 		"idle-consumer-limit", "idle-task-execution-limit",
 		"cache-level", "subscription-durable", "durable-subscription-name",
+		"subscription-shared", "subscription-name",
 		"client-id", "task-executor"
 	};
 
@@ -167,6 +168,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, pubSubDomainAttribute, "pubSubDomain");
 		}
 
+		if (StringUtils.hasText(element.getAttribute("subsription-name"))
+				&& StringUtils.hasText(element.getAttribute("durable-subsription-name"))) {
+			parserContext.getReaderContext().error(
+					"Only one of 'subscription-name' or 'durable-subscription-name' is allowed.", element);
+		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
@@ -182,6 +188,8 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "subscription-durable");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "durable-subscription-name",
 				"subscriptionName");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "subscription-shared");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "subscription-name");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id");
 		builder.addPropertyValue("autoStartup", false);
 		String beanName = adapterBeanNameRoot(element, parserContext, adapterBeanDefinition)

spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.2.xsd

@@ -97,10 +97,21 @@
 					<xsd:attribute name="subscription" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation><![CDATA[
-	The name for the durable subscription, if any.
+	The name for the durable or shared subscription, if any.
 							]]></xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="subscription-shared" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Boolean property indicating whether to make the subscription shared. The shared subscription name to be used can be
+	specified through the 'subscription' attribute. Default is "false". Set this to "true" to register a shared
+	subscription.
+	Note that shared subscriptions may also be durable, so this flag can
+	(and often will) be combined with 'durable' as well. Requires a JMS 2.0 compatible message broker.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -376,9 +387,10 @@
 					<xsd:attribute name="pub-sub-domain" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
-	The boolean property used to configure the JmsTemplate with knowledge of what JMS domain is being used. By default the value of this property is 'false'',
-	indicating that the point-to-point domain, Queues, will be used. This property used by JmsTemplate determines the behavior of dynamic destination resolution
-	via implementations of the DestinationResolver interface.
+	The boolean property used to configure the listener container with knowledge of what JMS domain is being used.
+	Unless 'subscription-shared' or 'subscription-durable' is true, by default the value of this property is 'false' ,
+	indicating that the point-to-point domain, Queues, will be used. When true, the pub/sub domain, Topics, will
+	be used.
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
@@ -395,13 +407,38 @@
 					<xsd:attribute name="durable-subscription-name" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
+	[DEPRECATED: Use 'subscription-durable="true"' together with 'subscription-name'].
 	The name of a durable subscription to create. To be applied in case of a topic (pub-sub domain) with subscription durability
 	activated. The durable subscription name needs to be unique within this client's JMS client id. Default is the class name of the
-	specified message listener. Note: Only 1 concurrent consumer (which is the default of this message listener container) is allowed
+	specified message listener. Note: Only 1 concurrent consumer (which is the default of the message listener container) is allowed
 	for each durable subscription.
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="subscription-shared" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Boolean property indicating whether to make the subscription shared. The shared subscription name to be used can be
+	specified through the 'subscription-name' attribute. Default is "false". Set this to "true" to register a shared
+	subscription, typically in combination with a 'subscription-name' value (unless your message listener class name is
+	good enough as subscription name). Note that shared subscriptions may also be durable, so this flag can
+	(and often will) be combined with 'subscription-durable' as well.
+	Only makes sense when listening to a topic (pub-sub domain), therefore this automatically sets
+	'pub-sub-domain' to 'true'. Requires a JMS 2.0 compatible message broker.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="subscription-name" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	The name of a subscription to create. To be applied in case of a topic (pub-sub domain) with a shared or durable
+	subscription. The subscription name needs to be unique within this client's JMS client id. Default is the class name
+	of the specified message listener.
+	Note: Only 1 concurrent consumer (which is the default of the message listener container) is allowed for each
+	subscription, except for a shared subscription (which requires JMS 2.0).
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="client-id" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
@@ -686,13 +723,38 @@
 					<xsd:attribute name="durable-subscription-name" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>
+	[DEPRECATED: Use 'subscription-durable="true"' together with 'subscription-name'].
 	The name of a durable subscription to create. To be applied in case of a topic (pub-sub domain) with subscription durability
 	activated. The durable subscription name needs to be unique within this client's JMS client id. Default is the class name of the
-	specified message listener. Note: Only 1 concurrent consumer (which is the default of this message listener container) is allowed
+	specified message listener. Note: Only 1 concurrent consumer (which is the default of the message listener container) is allowed
 	for each durable subscription.
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="subscription-shared" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Boolean property indicating whether to make the subscription shared. The shared subscription name to be used can be
+	specified through the 'subscription-name' attribute. Default is "false". Set this to "true" to register a shared
+	subscription, typically in combination with a 'subscription-name' value (unless your message listener class name is
+	good enough as subscription name). Note that shared subscriptions may also be durable, so this flag can
+	(and often will) be combined with 'subscription-durable' as well.
+	Only makes sense when listening to a topic (pub-sub domain), therefore this automatically sets
+	'request-pub-sub-domain' to 'true'. Requires a JMS 2.0 compatible message broker.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="subscription-name" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	The name of a subscription to create. To be applied in case of a topic (pub-sub domain) with a shared or durable
+	subscription. The subscription name needs to be unique within this client's JMS client id. Default is the class name
+	of the specified message listener.
+	Note: Only 1 concurrent consumer (which is the default of the message listener container) is allowed for each
+	subscription, except for a shared subscription (which requires JMS 2.0).
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="client-id" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation>

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml

@@ -23,7 +23,9 @@
 
 	<jms:publish-subscribe-channel id="topicReferenceChannel" topic="testTopic"/>
 
-	<jms:publish-subscribe-channel id="topicNameChannel" topic-name="test.topic"/>
+	<jms:publish-subscribe-channel id="topicNameChannel" topic-name="test.topic"
+		auto-startup="false"
+		durable="true" subscription="subName" subscription-shared="true" />
 
 	<jms:publish-subscribe-channel id="topicNameWithResolverChannel" topic-name="foo"
 			     destination-resolver="destinationResolver" connection-factory="connFact"/>

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.jms.config;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 import java.util.List;
 
@@ -185,6 +186,9 @@ public class JmsChannelParserTests {
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container");
 		assertEquals("test.topic", jmsTemplate.getDefaultDestinationName());
 		assertEquals("test.topic", container.getDestinationName());
+		assertTrue(container.isSubscriptionShared());
+		assertTrue(container.isSubscriptionDurable());
+		assertEquals("subName", container.getSubscriptionName());
 	}
 
 	@Test

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,10 @@
 
 package org.springframework.integration.jms.config;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Properties;
 
@@ -42,6 +45,7 @@ import org.springframework.messaging.PollableChannel;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class JmsInboundGatewayParserTests {
 
@@ -61,7 +65,7 @@ public class JmsInboundGatewayParserTests {
 		assertEquals("jms:inbound-gateway", componentHistoryRecord.get("type"));
 		assertNotNull("message should not be null", message);
 		assertEquals("message-driven-test", message.getPayload());
-		context.stop();
+		context.close();
 	}
 
 	@Test
@@ -75,7 +79,7 @@ public class JmsInboundGatewayParserTests {
 		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
 		assertEquals("message-driven-test", message.getPayload());
-		context.stop();
+		context.close();
 	}
 
 	@Test
@@ -89,7 +93,7 @@ public class JmsInboundGatewayParserTests {
 		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
 		assertEquals("converted-test-message", message.getPayload());
-		context.stop();
+		context.close();
 	}
 
 	@Test
@@ -100,6 +104,7 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(gateway);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("listener"));
 		assertEquals(Boolean.TRUE, accessor.getPropertyValue("extractReplyPayload"));
+		context.close();
 	}
 
 	@Test
@@ -110,6 +115,7 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(gateway);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("listener"));
 		assertEquals(Boolean.TRUE, accessor.getPropertyValue("extractReplyPayload"));
+		context.close();
 	}
 
 	@Test
@@ -120,6 +126,7 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(gateway);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("listener"));
 		assertEquals(Boolean.FALSE, accessor.getPropertyValue("extractReplyPayload"));
+		context.close();
 	}
 
 	@Test
@@ -130,6 +137,7 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(gateway);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("listener"));
 		assertEquals(Boolean.TRUE, accessor.getPropertyValue("extractRequestPayload"));
+		context.close();
 	}
 
 	@Test
@@ -140,12 +148,13 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(gateway);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("listener"));
 		assertEquals(Boolean.FALSE, accessor.getPropertyValue("extractRequestPayload"));
+		context.close();
 	}
 
 	@Test(expected = BeanDefinitionStoreException.class)
 	public void testGatewayWithConnectionFactoryOnly() {
 		try {
-			new ClassPathXmlApplicationContext("jmsGatewayWithConnectionFactoryOnly.xml", this.getClass());
+			new ClassPathXmlApplicationContext("jmsGatewayWithConnectionFactoryOnly.xml", this.getClass()).close();
 		}
 		catch (BeanDefinitionStoreException e) {
 			assertTrue(e.getMessage().contains("request-destination"));
@@ -157,7 +166,7 @@ public class JmsInboundGatewayParserTests {
 	@Test(expected = BeanDefinitionStoreException.class)
 	public void testGatewayWithEmptyConnectionFactory() {
 		try {
-			new ClassPathXmlApplicationContext("jmsGatewayWithEmptyConnectionFactory.xml", this.getClass());
+			new ClassPathXmlApplicationContext("jmsGatewayWithEmptyConnectionFactory.xml", this.getClass()).close();
 		}
 		catch (BeanDefinitionStoreException e) {
 			assertTrue(e.getMessage().contains("connection-factory"));
@@ -176,7 +185,7 @@ public class JmsInboundGatewayParserTests {
 		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
 		assertEquals("message-driven-test", message.getPayload());
-		context.stop();
+		context.close();
 	}
 
 	@Test
@@ -187,6 +196,7 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(gateway);
 		accessor = new DirectFieldAccessor(accessor.getPropertyValue("listenerContainer"));
 		assertNull(accessor.getPropertyValue("transactionManager"));
+		context.close();
 	}
 
 	@Test
@@ -200,6 +210,7 @@ public class JmsInboundGatewayParserTests {
 		assertEquals(JmsTransactionManager.class, txManager.getClass());
 		assertEquals(context.getBean("txManager"), txManager);
 		assertEquals(context.getBean("testConnectionFactory"), ((JmsTransactionManager) txManager).getConnectionFactory());
+		context.close();
 	}
 
 	@Test
@@ -212,6 +223,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("concurrentConsumers"));
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -224,6 +236,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(22, new DirectFieldAccessor(container).getPropertyValue("maxConcurrentConsumers"));
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -236,6 +249,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(99, new DirectFieldAccessor(container).getPropertyValue("maxMessagesPerTask"));
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -248,6 +262,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(1111L, new DirectFieldAccessor(container).getPropertyValue("receiveTimeout"));
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -267,6 +282,7 @@ public class JmsInboundGatewayParserTests {
 		}
 		assertEquals(2222L, recoveryInterval);
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -279,6 +295,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(7, new DirectFieldAccessor(container).getPropertyValue("idleTaskExecutionLimit"));
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -291,6 +308,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit"));
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -303,6 +321,7 @@ public class JmsInboundGatewayParserTests {
 				new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer");
 		assertEquals(context.getBean("messageListenerContainer"), container);
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -316,6 +335,7 @@ public class JmsInboundGatewayParserTests {
 		String messageSelector = (String) new DirectFieldAccessor(container).getPropertyValue("messageSelector");
 		assertEquals("TestProperty = 'foo'", messageSelector);
 		gateway.stop();
+		context.close();
 	}
 
 	@Test
@@ -329,6 +349,7 @@ public class JmsInboundGatewayParserTests {
 		assertEquals(7, accessor.getPropertyValue("replyPriority"));
 		assertEquals(DeliveryMode.NON_PERSISTENT, accessor.getPropertyValue("replyDeliveryMode"));
 		assertEquals(true, accessor.getPropertyValue("explicitQosEnabledForReplies"));
+		context.close();
 	}
 
 	@Test
@@ -339,6 +360,7 @@ public class JmsInboundGatewayParserTests {
 		DirectFieldAccessor accessor = new DirectFieldAccessor(
 				new DirectFieldAccessor(gateway).getPropertyValue("listener"));
 		assertEquals(false, accessor.getPropertyValue("explicitQosEnabledForReplies"));
+		context.close();
 	}
 
 	@Test
@@ -348,6 +370,7 @@ public class JmsInboundGatewayParserTests {
 		JmsMessageDrivenEndpoint endpoint = context.getBean("gateway", JmsMessageDrivenEndpoint.class);
 		JmsDestinationAccessor container = (JmsDestinationAccessor) new DirectFieldAccessor(endpoint).getPropertyValue("listenerContainer");
 		assertEquals(Boolean.TRUE, container.isPubSubDomain());
+		context.close();
 	}
 
 	@Test
@@ -360,6 +383,8 @@ public class JmsInboundGatewayParserTests {
 		assertEquals(Boolean.TRUE, container.isSubscriptionDurable());
 		assertEquals("testDurableSubscriptionName", container.getDurableSubscriptionName());
 		assertEquals("testClientId", container.getClientId());
+		assertTrue(container.isSubscriptionShared());
+		context.close();
 	}
 
 	@Test
@@ -373,6 +398,7 @@ public class JmsInboundGatewayParserTests {
 		JmsTemplate template = new JmsTemplate(context.getBean(ConnectionFactory.class));
 		template.convertAndSend("testDestination", "Hello");
 		assertNotNull(template.receive("testReplyDestination"));
+		context.close();
 	}
 
 }

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java

@@ -21,6 +21,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Properties;
 
@@ -88,6 +89,7 @@ public class JmsMessageDrivenChannelAdapterParserTests {
 		assertEquals(Boolean.TRUE, container.isSubscriptionDurable());
 		assertEquals("testDurableSubscriptionName", container.getDurableSubscriptionName());
 		assertEquals("testClientId", container.getClientId());
+		assertTrue(container.isSubscriptionShared());
 		endpoint.stop();
 		context.close();
 	}

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithDurableSubscription.xml

@@ -15,11 +15,13 @@
 	</si:channel>
 
 	<jms:inbound-gateway id="gateway"
+						 auto-startup="false"
 					 	 request-destination-name="testDestinationName"
 					 	 request-channel="requestChannel"
 					 	 request-pub-sub-domain="true"
 					 	 subscription-durable="true"
-						 durable-subscription-name="testDurableSubscriptionName"
+						 subscription-name="testDurableSubscriptionName"
+						 subscription-shared="true"
 						 client-id="testClientId" />
 
 	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithDurableSubscription.xml

@@ -15,10 +15,12 @@
 	</si:channel>
 
 	<jms:message-driven-channel-adapter id="messageDrivenAdapter"
+			auto-startup="false"
 			destination-name="testDestination"
 			pub-sub-domain="true"
 			subscription-durable="true"
-			durable-subscription-name="testDurableSubscriptionName"
+			subscription-name="testDurableSubscriptionName"
+			subscription-shared="true"
 			client-id="testClientId"
 			channel="output"/>
 

src/reference/asciidoc/jms.adoc

@@ -106,6 +106,11 @@ When comparing this to the generic gateway configuration, or the JMS 'inbound-ga
 Therefore, the flow downstream from the 'error-channel' should also be one-way.
 For example, it could simply send to a logging handler, or it could be connected to a different JMS <outbound-channel-adapter> element.
 
+When consuming from topics, set the `pub-sub-domain` attribute to true; set `subscription-durable` to true
+for a durable subscription, `subscription-shared` for a shared subscription (requires a JMS 2.0 broker and
+has been available since _version 4.2_).
+Use `subscription-name` to name the subscription.
+
 [[jms-md-conversion-errors]]
 ==== Inbound Conversion Errors
 [NOTE]
@@ -196,6 +201,11 @@ The same idea applies here: The _exceptionTransformer_ could be a simple POJO th
 
 NOTE: See <<jms-md-conversion-errors>>.
 
+When consuming from topics, set the `pub-sub-domain` attribute to true; set `subscription-durable` to true
+for a durable subscription, `subscription-shared` for a shared subscription (requires a JMS 2.0 broker and
+has been available since _version 4.2_).
+Use `subscription-name` to name the subscription.
+
 [[jms-outbound-gateway]]
 === Outbound Gateway
 
@@ -494,6 +504,11 @@ The example below provides both a custom instance for resolution of the JMS Dest
     connection-factory="customConnectionFactory"/>
 ----
 
+For the `<publish-subscribe-channel />`; set the `durable` attribute to true
+for a durable subscription, `subscription-shared` for a shared subscription (requires a JMS 2.0 broker and
+has been available since _version 4.2_).
+Use `subscription` to name the subscription.
+
 [[jms-selectors]]
 === Using JMS Message Selectors
 

src/reference/asciidoc/whats-new.adoc

@@ -155,9 +155,19 @@ Of course, Reactor is still supported for functionality such as the `Promise` ga
 [[x4.2-jms-changes]]
 ==== JMS Changes
 
+===== Conversion Errors in Message-Driven Endpoints
+
 The `error-channel` now is used for the conversion errors, which have caused a transaction rollback and message redelivery previously.
 
-See <<jms-message-driven-channel-adapter>> for more information.
+See <<jms-message-driven-channel-adapter>> and <<jms-inbound-gateway>> for more information.
+
+===== Shared Subscriptions
+
+Namespace support for shared subscriptions (JMS 2.0) has been added to message-driven endpoints and the
+`<int-jms:publish-subscribe-channel>`.
+Previously, you had to wire up listener containers as `<bean/>` s to use shared connections.
+
+See <<jms>> for more information.
 
 [[x4.2-conditional-pollers]]
 ==== Conditional Pollers
@@ -194,7 +204,7 @@ See <<cors>> for more information.
 ==== Persistent File List Filter Changes
 
 The `AbstractPersistentFileListFilter` has a new property `flushOnUpdate` which, when set to true, will `flush()` the
-metadata store if it implements `Flushable` (e.g. the `PropertiesPersistenMetadataStore`).
+metadata store if it implements `Flushable` (e.g. the `PropertiesPersistingMetadataStore`).
 
 
 [[x4.2-gw]]
@@ -223,7 +233,7 @@ Currently, JDBC, Redis and MongoDB message stores support this property.
 
 ===== Output MessageGroupProcessor
 
-When using a `ref` or innner bean for the aggregator, it is now possible to bind a `MessageGroupProcessor` directly.
+When using a `ref` or inner bean for the aggregator, it is now possible to bind a `MessageGroupProcessor` directly.
 In addition, a `SimpleMessageGroupProcessor` is provided that simply returns the collection of messages in the group.
 When an output processor produces a collection of `Message<?>`, the aggregator releases those messages individually.
 Configuring the `SimpleMessageGroupProcessor` makes the aggregator a message barrier, were messages are held up