Merge pull request #819 from garyrussell/INT-3053

* INT-3053: Allow task-executor on <reply-listener/>
2013-06-09 20:37:43 -04:00
parent 1a27d81941 6d4f2b46d8
commit cc5bcf64a6
7 changed files with 109 additions and 0 deletions
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
@@ -23,6 +23,7 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -59,6 +60,7 @@ import org.springframework.jms.support.converter.SimpleMessageConverter;
 import org.springframework.jms.support.destination.DestinationResolver;
 import org.springframework.jms.support.destination.DynamicDestinationResolver;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;

 /**
@@ -553,6 +555,17 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 			if (this.replyContainerProperties.getSessionAcknowledgeMode() != null) {
 				container.setSessionAcknowledgeMode(this.replyContainerProperties.getSessionAcknowledgeMode());
 			}
+			if (this.replyContainerProperties.getTaskExecutor() != null) {
+				container.setTaskExecutor(this.replyContainerProperties.getTaskExecutor());
+			}
+			else {
+				// set the beanName so the default TE threads get a meaningful name
+				String containerBeanName = this.getComponentName();
+				containerBeanName = ((!StringUtils.hasText(containerBeanName)
+						? "JMS_OutboundGateway@" + ObjectUtils.getIdentityHexString(this)
+						: containerBeanName) + ".replyListener");
+				container.setBeanName(containerBeanName);
+			}
 		}
 	}

@@ -1133,6 +1146,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp

 		private volatile Integer idleTaskExecutionLimit;

+		private volatile Executor taskExecutor;
+
 		public Boolean isSessionTransacted() {
 			return sessionTransacted;
 		}
@@ -1212,5 +1227,13 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 		public void setIdleTaskExecutionLimit(Integer idleTaskExecutionLimit) {
 			this.idleTaskExecutionLimit = idleTaskExecutionLimit;
 		}
+
+		public void setTaskExecutor(Executor taskExecutor) {
+			this.taskExecutor = taskExecutor;
+		}
+
+		public Executor getTaskExecutor() {
+			return taskExecutor;
+		}
 	}
 }
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java
@@ -24,6 +24,7 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.jms.JmsOutboundGateway;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
+
 import org.w3c.dom.Element;

 /**
@@ -133,6 +134,7 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-consumer-limit");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-task-execution-limit");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "cache-level");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");

 		gatewayBuilder.addPropertyValue("replyContainerProperties", builder.getBeanDefinition());
 	}
--- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd
+++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd
@@ -760,6 +760,20 @@
 						</xsd:attribute>
 						<xsd:attributeGroup ref="dmlcAttributeGroup" />
 						<xsd:attribute name="receive-timeout" type="xsd:string"/>
+						<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 receiving
+		the replies and handing them over to the sending thread.
+		Default is a SimpleAsyncTaskExecutor.
+								]]></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:complexType>
 				</xsd:element>
 			</xsd:choice>
--- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd
+++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd
@@ -760,6 +760,20 @@
 						</xsd:attribute>
 						<xsd:attributeGroup ref="dmlcAttributeGroup" />
 						<xsd:attribute name="receive-timeout" type="xsd:string"/>
+						<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 receiving
+		the replies and handing them over to the sending thread.
+		Default is a SimpleAsyncTaskExecutor.
+								]]></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:complexType>
 				</xsd:element>
 			</xsd:choice>
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2002-2013 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 static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import javax.jms.ConnectionFactory;
+
+import org.springframework.integration.jms.JmsOutboundGateway.ReplyContainerProperties;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.util.ObjectUtils;
+
+import org.junit.Test;
+
+/**
+ * @author Gary Russell
+ * @since 2.2.4
+ *
+ */
+public class JmsOutboundGatewayTests {
+
+	@Test
+	public void testContainerBeanNameWhenNoGatewayBeanName() {
+		JmsOutboundGateway gateway = new JmsOutboundGateway();
+		gateway.setConnectionFactory(mock(ConnectionFactory.class));
+		gateway.setRequestDestinationName("foo");
+		gateway.setUseReplyContainer(true);
+		gateway.setReplyContainerProperties(new ReplyContainerProperties());
+		gateway.afterPropertiesSet();
+		assertEquals("JMS_OutboundGateway@" + ObjectUtils.getIdentityHexString(gateway) +
+				".replyListener",
+			TestUtils.getPropertyValue(gateway, "replyContainer.beanName"));
+	}
+}
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java
@@ -38,6 +38,7 @@ import javax.jms.Session;

 import org.junit.Test;
 import org.mockito.Mockito;
+
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -92,6 +93,7 @@ public class JmsOutboundGatewayParserTests {
 		assertEquals(2, TestUtils.getPropertyValue(container, "idleTaskExecutionLimit"));
 		assertEquals(3, TestUtils.getPropertyValue(container, "cacheLevel"));
 		assertTrue(container.isSessionTransacted());
+		assertSame(context.getBean("exec"), TestUtils.getPropertyValue(container, "taskExecutor"));
 	}

 	@Test
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml
@@ -3,10 +3,13 @@
 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xmlns:si="http://www.springframework.org/schema/integration"
 	xmlns:jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:task="http://www.springframework.org/schema/task"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/task
+			http://www.springframework.org/schema/task/spring-task.xsd
 			http://www.springframework.org/schema/integration/jms
 			http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">

@@ -26,9 +29,12 @@
 			recovery-interval="10000"
 			idle-consumer-limit="7"
 			idle-task-execution-limit="2"
+			task-executor="exec"
 			cache-level="3" />
 	</jms:outbound-gateway>

+	<task:executor id="exec" />
+
 	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		<constructor-arg>
 			<bean class="org.springframework.integration.jms.StubConnection">