From d19f126da946ae96b920acc362af7ef022c79676 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 8 Mar 2011 18:27:02 -0500
Subject: [PATCH] INT-1830 added support for resolution of 'concurrency' with
 property placeholder

---
 .../jms/config/JmsChannelFactoryBean.java     | 64 +++++++++----------
 .../jms/PollableJmsChannelTests.java          | 19 +++---
 .../config/JmsChannelParserTests-context.xml  |  3 +-
 3 files changed, 40 insertions(+), 46 deletions(-)

diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
index 12c583af38..7c13f335c1 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
@@ -73,7 +73,8 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 	private volatile String clientId;
 
 	private volatile String concurrency;
-	//private volatile Integer concurrentConsumers;
+	
+	private volatile Integer concurrentConsumers;
 
 	private volatile ConnectionFactory connectionFactory;
 
@@ -93,7 +94,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 
 	private volatile Integer idleTaskExecutionLimit;
 
-	//private volatile Integer maxConcurrentConsumers;
+	private volatile Integer maxConcurrentConsumers;
 
 	private volatile Integer maxMessagesPerTask;
 
@@ -200,9 +201,10 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 	public void setConcurrency(String concurrency) {
 		this.concurrency = concurrency;
 	}
-//	public void setConcurrentConsumers(int concurrentConsumers) {
-//		this.concurrentConsumers = concurrentConsumers;
-//	}
+	
+	public void setConcurrentConsumers(int concurrentConsumers) {
+		this.concurrentConsumers = concurrentConsumers;
+	}
 
 	public void setConnectionFactory(ConnectionFactory connectionFactory) {
 		this.connectionFactory = connectionFactory;
@@ -246,9 +248,9 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		this.idleTaskExecutionLimit = idleTaskExecutionLimit;
 	}
 
-//	public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
-//		this.maxConcurrentConsumers = maxConcurrentConsumers;
-//	}
+	public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
+		this.maxConcurrentConsumers = maxConcurrentConsumers;
+	}
 
 	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
 		this.maxMessagesPerTask = maxMessagesPerTask;
@@ -386,23 +388,26 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		container.setSessionTransacted(this.sessionTransacted);
 		container.setSubscriptionDurable(this.subscriptionDurable);
 		
-		int[] conc = parseConcurrency(concurrency);
+		
 		
 		if (container instanceof DefaultMessageListenerContainer) {
 			DefaultMessageListenerContainer dmlc = (DefaultMessageListenerContainer) container;
 			if (this.cacheLevelName != null) {
 				dmlc.setCacheLevelName(this.cacheLevelName);
 			}
-					
-			if (conc != null) {
-				if (containerType.isAssignableFrom(DefaultMessageListenerContainer.class)) {
-					dmlc.setConcurrentConsumers(conc[0]);
-					if (conc.length == 2){
-						dmlc.setMaxConcurrentConsumers(conc[1]);
-					}			
-				}
+		
+			if (StringUtils.hasText(this.concurrency)){
+				dmlc.setConcurrency(this.concurrency);
 			}
 			
+			if (this.concurrentConsumers != null){
+				dmlc.setConcurrentConsumers(this.concurrentConsumers);
+			}
+			
+			if (this.maxConcurrentConsumers != null){
+				dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
+			}
+				
 			if (this.idleTaskExecutionLimit != null) {
 				dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit);
 			}
@@ -428,9 +433,14 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		}
 		else if (container instanceof SimpleMessageListenerContainer) {
 			SimpleMessageListenerContainer smlc = (SimpleMessageListenerContainer) container;
-			if (conc != null) {
-				smlc.setConcurrentConsumers(conc[0]);
+			if (StringUtils.hasText(this.concurrency)){
+				smlc.setConcurrency(this.concurrency);
 			}
+			
+			if (this.concurrentConsumers != null){
+				smlc.setConcurrentConsumers(this.concurrentConsumers);
+			}
+			
 			smlc.setPubSubNoLocal(this.pubSubNoLocal);
 			smlc.setTaskExecutor(this.taskExecutor);
 		}
@@ -479,20 +489,4 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 			((SubscribableJmsChannel) this.channel).destroy();
 		}
 	}
-	
-	private int[] parseConcurrency(String concurrency) {
-		if (!StringUtils.hasText(concurrency)) {
-			return null;
-		}
-		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)};
-		}
-	}
 }
diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java
index 39645093af..9569f57d1e 100644
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java
@@ -24,7 +24,6 @@ import javax.jms.Destination;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.command.ActiveMQQueue;
-import org.junit.Before;
 import org.junit.Test;
 
 import org.springframework.integration.Message;
@@ -41,17 +40,13 @@ public class PollableJmsChannelTests {
 
 	private Destination queue;
 
-
-	@Before
-	public void setup() throws Exception {
+	@Test
+	public void queueReference() throws Exception {
 		ActiveMqTestUtils.prepare();
 		this.connectionFactory = new ActiveMQConnectionFactory();
 		this.connectionFactory.setBrokerURL("vm://localhost");
 		this.queue = new ActiveMQQueue("pollableJmsChannelTestQueue");
-	}
-
-	@Test
-	public void queueReference() throws Exception {
+		
 		JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
 		factoryBean.setConnectionFactory(this.connectionFactory);
 		factoryBean.setDestination(this.queue);
@@ -71,9 +66,13 @@ public class PollableJmsChannelTests {
 
 	@Test
 	public void queueName() throws Exception {
+		ActiveMqTestUtils.prepare();
+		this.connectionFactory = new ActiveMQConnectionFactory();
+		this.connectionFactory.setBrokerURL("vm://localhost");
+		
 		JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
 		factoryBean.setConnectionFactory(this.connectionFactory);
-		factoryBean.setDestinationName("dynamicQueue");
+		factoryBean.setDestinationName("someDynamicQueue");
 		factoryBean.setPubSubDomain(false);
 		factoryBean.afterPropertiesSet();
 		PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
@@ -81,7 +80,7 @@ public class PollableJmsChannelTests {
 		assertTrue(sent1);
 		boolean sent2 = channel.send(new GenericMessage<String>("bar"));
 		assertTrue(sent2);
-		Message<?> result1 = channel.receive(1000);
+		Message<?> result1 = channel.receive(10000);
 		assertNotNull(result1);
 		assertEquals("foo", result1.getPayload());
 		Message<?> result2 = channel.receive(1000);
diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml
index 1caed9d6eb..b041350831 100644
--- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml
+++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml
@@ -61,6 +61,7 @@
 	<alias name="connectionFactory" alias="connFact"/>
 	
 	
-	<jms:channel id="withPlaceholders" queue="${queue}" concurrency="${concurrency}"/>
+	<jms:channel id="withPlaceholders" queue="${queue}" 
+									   concurrency="${concurrency}"/>
 
 </beans>