INT-1830 added support for resolution of 'concurrency' with property placeholder
This commit is contained in:
@@ -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)};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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>
|
||||
|
||||
Reference in New Issue
Block a user