diff --git 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 index 02513da4e0..ba4f243104 100644 --- 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; + } } } diff --git 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 index e4eda32e56..96ec1cb690 100644 --- 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()); } diff --git 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 index f6b12aa627..5e16ef13e4 100644 --- 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 @@ + + + + + + + + + + diff --git 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 index 07ba7f27de..21b5f39459 100644 --- 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 @@ + + + + + + + + + + diff --git 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 new file mode 100644 index 0000000000..c8099582a9 --- /dev/null +++ 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")); + } +} diff --git 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 index 2c3a765e43..31762490c6 100644 --- 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 diff --git 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 index fda5a06b97..f406f83c19 100644 --- 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" /> + +