Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove overridden DMLC methods accepting Queue

Browse Source 
* As the implementations of those in `AbstractMessageListenerContainer`
delegate to ones accepting `String`s.

* Set `GRADLE_OPTS` in travis
It failed on Travis sporadically with some "killed" message, sounds like main gradle process used too much memory.

* Extract `collectQueueNames(Queue... queues)`

* Make `Queue`-based methods in `AbstractMessageListenerContainer` as `final`

* Remove "this." from method calls

* Add `@author` to affected classes
This commit is contained in:
Alex Panchenko
2017-03-14 20:40:36 +07:00
committed by Artem Bilan
parent d7e702a765
commit eeab98e365
6 changed files with 67 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.travis.yml
View File

@@ -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

40
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -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 <code>null</code>)
*/
public void setQueues(Queue... queues) {
List<String> queueNames = new ArrayList<String>(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));
}
/**

29
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
View File

@@ -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<String> 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<String> 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);
}

16
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java
View File

@@ -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(

39
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -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.

45
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java
View File

@@ -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<String, String>) 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");