diff --git a/.travis.yml b/.travis.yml index de0f7f38..9cb2eed8 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,6 +4,8 @@ sudo: false services: - rabbitmq env: + global: + - GRADLE_OPTS="-Xmx512m" - TERM=dumb before_cache: - rm -f $HOME/.gradle/caches/modules-2/modules-2.lock diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index d3668d5d..86d7515b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -17,7 +17,6 @@ package org.springframework.amqp.rabbit.listener; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.HashMap; @@ -89,6 +88,7 @@ import com.rabbitmq.client.ShutdownSignalException; * @author Dave Syer * @author James Carr * @author Gary Russell + * @author Alex Panchenko */ public abstract class AbstractMessageListenerContainer extends RabbitAccessor implements MessageListenerContainer, ApplicationContextAware, BeanNameAware, DisposableBean, @@ -245,21 +245,25 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor */ public void setQueueNames(String... queueName) { Assert.noNullElements(queueName, "Queue name(s) cannot be null"); - this.queueNames = new CopyOnWriteArrayList<>(Arrays.asList(queueName)); + this.queueNames = new CopyOnWriteArrayList<>(queueName); } /** * Set the name of the queue(s) to receive messages from. * @param queues the desired queue(s) (can not be null) */ - public void setQueues(Queue... queues) { - List queueNames = new ArrayList(queues.length); + public final void setQueues(Queue... queues) { + setQueueNames(collectQueueNames(queues)); + } + + private static String[] collectQueueNames(Queue... queues) { + Assert.notNull(queues, "'queues' cannot be null"); + Assert.noNullElements(queues, "'queues' cannot contain null elements"); + String[] queueNames = new String[queues.length]; for (int i = 0; i < queues.length; i++) { - Assert.notNull(queues[i], "Queue (" + i + ") must not be null."); - queueNames.add(queues[i].getName()); + queueNames[i] = queues[i].getName(); } - queueNames = new CopyOnWriteArrayList<>(queueNames); - this.queueNames = queueNames; + return queueNames; } /** @@ -292,14 +296,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * Add queue(s) to this container's list of queues. * @param queues The queue(s) to add. */ - public void addQueues(Queue... queues) { - Assert.notNull(queues, "'queues' cannot be null"); - Assert.noNullElements(queues, "'queues' cannot contain null elements"); - String[] queueNames = new String[queues.length]; - for (int i = 0; i < queues.length; i++) { - queueNames[i] = queues[i].getName(); - } - this.addQueueNames(queueNames); + public final void addQueues(Queue... queues) { + addQueueNames(collectQueueNames(queues)); } /** @@ -318,14 +316,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * @param queues The queue(s) to remove. * @return the boolean result of removal on the target {@code queueNames} List. */ - public boolean removeQueues(Queue... queues) { - Assert.notNull(queues, "'queues' cannot be null"); - Assert.noNullElements(queues, "'queues' cannot contain null elements"); - String[] queueNames = new String[queues.length]; - for (int i = 0; i < queues.length; i++) { - queueNames[i] = queues[i].getName(); - } - return this.removeQueueNames(queueNames); + public final boolean removeQueues(Queue... queues) { + return removeQueueNames(collectQueueNames(queues)); } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index 24931f15..93cfab25 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -24,6 +24,7 @@ import java.util.Collections; import java.util.Date; import java.util.LinkedList; import java.util.List; +import java.util.Objects; import java.util.Properties; import java.util.Set; import java.util.concurrent.CountDownLatch; @@ -41,7 +42,6 @@ import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; -import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; @@ -195,12 +195,6 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta super.setMissingQueuesFatal(missingQueuesFatal); } - @Override - public void setQueues(Queue... queues) { - Assert.state(!isRunning(), "Cannot set queue names while running, use add/remove"); - super.setQueues(queues); - } - @Override public void addQueueNames(String... queueNames) { try { @@ -212,18 +206,6 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta super.addQueueNames(queueNames); } - @Override - public void addQueues(Queue... queues) { - try { - addQueues(Arrays.stream(queues) - .map(Queue::getName)); - } - catch (AmqpIOException e) { - throw new AmqpIOException("Failed to add " + Arrays.asList(queues), e.getCause()); - } - super.addQueues(queues); - } - private void addQueues(Stream queueNameStream) { if (isRunning()) { synchronized (this.consumersMonitor) { @@ -248,19 +230,12 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta return super.removeQueueNames(queueNames); } - @Override - public boolean removeQueues(Queue... queues) { - removeQueues(Arrays.stream(queues) - .map(Queue::getName)); - return super.removeQueues(queues); - } - private void removeQueues(Stream queueNames) { if (isRunning()) { synchronized (this.consumersMonitor) { checkStartState(); queueNames.map(this.consumersByQueue::remove) - .filter(consumersOnQueue -> consumersOnQueue != null) + .filter(Objects::nonNull) .flatMap(Collection::stream) .forEach(this::cancelConsumer); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java index 6e21b5bc..8424ee6f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java @@ -22,7 +22,6 @@ import java.util.concurrent.ConcurrentMap; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.MessageListener; -import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.util.Assert; @@ -72,31 +71,16 @@ public class DirectReplyToMessageListenerContainer extends DirectMessageListener throw new UnsupportedOperationException(); } - @Override - public final void setQueues(Queue... queues) { - throw new UnsupportedOperationException(); - } - @Override public final void addQueueNames(String... queueNames) { throw new UnsupportedOperationException(); } - @Override - public final void addQueues(Queue... queues) { - throw new UnsupportedOperationException(); - } - @Override public final boolean removeQueueNames(String... queueNames) { throw new UnsupportedOperationException(); } - @Override - public final boolean removeQueues(Queue... queues) { - throw new UnsupportedOperationException(); - } - @Override public void setMessageListener(Object messageListener) { throw new UnsupportedOperationException( diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index d383639f..fca2ebc4 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -36,7 +36,6 @@ import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; -import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry; @@ -67,6 +66,8 @@ import com.rabbitmq.client.ShutdownSignalException; * @author Dave Syer * @author Gary Russell * @author Artem Bilan + * @author Alex Panchenko + * * @since 1.0 */ public class SimpleMessageListenerContainer extends AbstractMessageListenerContainer { @@ -302,12 +303,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.queuesChanged(); } - @Override - public void setQueues(Queue... queues) { - super.setQueues(queues); - this.queuesChanged(); - } - /** * Add queue(s) to this container's list of queues. The existing consumers * will be cancelled after they have processed any pre-fetched messages and @@ -321,19 +316,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.queuesChanged(); } - /** - * Add queue(s) to this container's list of queues. The existing consumers - * will be cancelled after they have processed any pre-fetched messages and - * new consumers will be created. The queue must exist to avoid problems when - * restarting the consumers. - * @param queue The queue to add. - */ - @Override - public void addQueues(Queue... queue) { - super.addQueues(queue); - this.queuesChanged(); - } - /** * Remove queues from this container's list of queues. The existing consumers * will be cancelled after they have processed any pre-fetched messages and @@ -351,23 +333,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } - /** - * Remove queue(s) from this container's list of queues. The existing consumers - * will be cancelled after they have processed any pre-fetched messages and - * new consumers will be created. At least one queue must remain. - * @param queue The queue to remove. - */ - @Override - public boolean removeQueues(Queue... queue) { - if (super.removeQueues(queue)) { - this.queuesChanged(); - return true; - } - else { - return false; - } - } - /** * Set the number of retries after passive queue declaration fails. * @param declarationRetries The number of retries, default 3. diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java index daa30722..dd966763 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.listener; import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -80,6 +81,8 @@ import com.rabbitmq.client.Consumer; /** * @author Gary Russell * @author Artem Bilan + * @author Alex Panchenko + * * @since 2.0 * */ @@ -223,6 +226,48 @@ public class DirectMessageListenerContainerIntegrationTests { cf.destroy(); } + @Test + public void testQueueManagementQueueInstances() throws Exception { + CachingConnectionFactory cf = new CachingConnectionFactory("localhost"); + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setThreadNamePrefix("client-"); + executor.afterPropertiesSet(); + cf.setExecutor(executor); + DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf); + container.setConsumersPerQueue(2); + container.setMessageListener(new MessageListenerAdapter((ReplyingMessageListener) in -> { + if ("foo".equals(in) || "bar".equals(in)) { + return in.toUpperCase(); + } + else { + return null; + } + })); + container.setBeanName("qManage"); + container.setConsumerTagStrategy(new Tag()); + container.afterPropertiesSet(); + container.setQueues(new Queue(Q1)); + assertArrayEquals(new String[] { Q1 }, container.getQueueNames()); + container.start(); + container.addQueues(new Queue(Q2)); + assertTrue(consumersOnQueue(Q1, 2)); + assertTrue(consumersOnQueue(Q2, 2)); + RabbitTemplate template = new RabbitTemplate(cf); + assertEquals("FOO", template.convertSendAndReceive(Q1, "foo")); + assertEquals("BAR", template.convertSendAndReceive(Q2, "bar")); + container.removeQueues(new Queue(Q1), new Queue(Q2), new Queue("junk")); + assertTrue(consumersOnQueue(Q1, 0)); + assertTrue(consumersOnQueue(Q2, 0)); + assertTrue(activeConsumerCount(container, 0)); + container.stop(); + assertTrue(consumersOnQueue(Q1, 0)); + assertTrue(consumersOnQueue(Q2, 0)); + assertTrue(activeConsumerCount(container, 0)); + assertEquals(0, TestUtils.getPropertyValue(container, "consumersByQueue", MultiValueMap.class).size()); + template.stop(); + cf.destroy(); + } + @Test public void testAddRemoveConsumers() throws Exception { CachingConnectionFactory cf = new CachingConnectionFactory("localhost");