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

GH-891 Introduced MultiRabbit to handle multiple brokers (#1111)

Browse Source 
* GH-891 Introduced MultiRabbit for handle multiple brokers

* GH-891 Fixed formatting and conventions

* GH-891 Avoid premature load of factory beans and some minor refactoring

* GH-891 Introduced tests for SpringMultirabbit

* GH-891 Modified BPP to return list of Declarables created

* GH-891 Minor refactoring

* GH-891 Updated copyright year

* GH-891 Introduced default setAdminsThatShouldDeclare() into Declarable

* GH-891 Avoiding premature instantiation of RabbitAdmin by providing the bean name instead

* GH-891 Changed to non-default

* GH-891 Removed MultiRabbitBootstrapConfiguration and its test
This commit is contained in:
Wander Costa
2020-06-24 23:01:32 +02:00
committed by GitHub
parent 4529429348
commit c7897ed977
11 changed files with 719 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java
View File

@@ -94,15 +94,7 @@ public abstract class AbstractDeclarable implements Declarable {
this.ignoreDeclarationExceptions = ignoreDeclarationExceptions;
}
/**
* The {@code AmqpAdmin}s that should declare this object; default is
* all admins.
* <br><br>A null argument, or an array/varArg with a single null argument, clears the collection
* ({@code setAdminsThatShouldDeclare((AmqpAdmin) null)} or
* {@code setAdminsThatShouldDeclare((AmqpAdmin[]) null)}). Clearing the collection resets
* the behavior such that all admins will declare the object.
* @param adminArgs The admins.
*/
@Override
public void setAdminsThatShouldDeclare(Object... adminArgs) {
Collection<Object> admins = new ArrayList<Object>();
if (adminArgs != null) {

11
spring-amqp/src/main/java/org/springframework/amqp/core/Declarable.java
View File

@@ -52,6 +52,17 @@ public interface Declarable {
*/
boolean isIgnoreDeclarationExceptions();
/**
* The {@code AmqpAdmin}s that should declare this object; default is
* all admins.
* <br><br>A null argument, or an array/varArg with a single null argument, clears the collection
* ({@code setAdminsThatShouldDeclare((AmqpAdmin) null)} or
* {@code setAdminsThatShouldDeclare((AmqpAdmin[]) null)}). Clearing the collection resets
* the behavior such that all admins will declare the object.
* @param adminArgs The admins.
*/
void setAdminsThatShouldDeclare(Object... adminArgs);
/**
* Add an argument to the declarable.
* @param name the argument name.

8
spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/RabbitListenerTestHarness.java
View File

@@ -16,6 +16,7 @@
package org.springframework.amqp.rabbit.test;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
@@ -30,6 +31,7 @@ import org.mockito.AdditionalAnswers;
import org.mockito.Mockito;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
@@ -73,8 +75,8 @@ public class RabbitListenerTestHarness extends RabbitListenerAnnotationBeanPostP
}
@Override
protected void processListener(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object bean,
Object target, String beanName) {
protected Collection<Declarable> processListener(MethodRabbitListenerEndpoint endpoint,
RabbitListener rabbitListener, Object bean, Object target, String beanName) {
Object proxy = bean;
String id = rabbitListener.id();
@@ -102,7 +104,7 @@ public class RabbitListenerTestHarness extends RabbitListenerAnnotationBeanPostP
else {
logger.info("The test harness can only proxy @RabbitListeners with an 'id' attribute");
}
super.processListener(endpoint, rabbitListener, proxy, target, beanName); // NOSONAR proxy is not null
return super.processListener(endpoint, rabbitListener, proxy, target, beanName); // NOSONAR proxy is not null
}
public InvocationData getNextInvocationDataFor(String id, long wait, TimeUnit unit) throws InterruptedException {

77
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/MultiRabbitListenerAnnotationBeanPostProcessor.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* Copyright 2020 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
*
* https://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.amqp.rabbit.annotation;
import java.lang.reflect.Method;
import java.util.Collection;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.util.StringUtils;
/**
* An extension of {@link RabbitListenerAnnotationBeanPostProcessor} that associates the
* proper RabbitAdmin to the beans of Exchanges, Queues, and Bindings after they are
* created.
* <p>
* This processing restricts the {@link RabbitAdmin} according to the related
* configuration, preventing the server from automatic binding non-related structures.
*
* @author Wander Costa
*/
public class MultiRabbitListenerAnnotationBeanPostProcessor extends RabbitListenerAnnotationBeanPostProcessor {
public static final String CONNECTION_FACTORY_BEAN_NAME = "multiRabbitConnectionFactory";
public static final String CONNECTION_FACTORY_CREATOR_BEAN_NAME = "rabbitConnectionFactoryCreator";
private static final String DEFAULT_RABBIT_ADMIN_BEAN_NAME = "defaultRabbitAdmin";
private static final String RABBIT_ADMIN_SUFFIX = "-admin";
@Override
protected Collection<Declarable> processAmqpListener(RabbitListener rabbitListener, Method method,
Object bean, String beanName) {
final Collection<Declarable> declarables = super.processAmqpListener(rabbitListener, method, bean, beanName);
final String rabbitAdmin = resolveMultiRabbitAdminName(rabbitListener);
for (final Declarable declarable : declarables) {
if (declarable.getDeclaringAdmins().isEmpty()) {
declarable.setAdminsThatShouldDeclare(rabbitAdmin);
}
}
return declarables;
}
/**
* Resolves the name of the RabbitAdmin bean based on the RabbitListener, or falls back to
* the default RabbitAdmin name provided by MultiRabbit.
*
* @param rabbitListener The RabbitListener to process the name from.
* @return The name of the RabbitAdmin bean.
*/
protected String resolveMultiRabbitAdminName(RabbitListener rabbitListener) {
String admin = super.resolveExpressionAsString(rabbitListener.admin(), "admin");
if (!StringUtils.hasText(admin) && StringUtils.hasText(rabbitListener.containerFactory())) {
admin = rabbitListener.containerFactory()
+ MultiRabbitListenerAnnotationBeanPostProcessor.RABBIT_ADMIN_SUFFIX;
}
if (!StringUtils.hasText(admin)) {
admin = MultiRabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_ADMIN_BEAN_NAME;
}
return admin;
}
}

41
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
View File

@@ -41,6 +41,7 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Base64UrlNamingStrategy;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Queue;
@@ -362,11 +363,12 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
}
protected void processAmqpListener(RabbitListener rabbitListener, Method method, Object bean, String beanName) {
protected Collection<Declarable> processAmqpListener(RabbitListener rabbitListener, Method method, Object bean,
String beanName) {
Method methodToUse = checkProxy(method, bean);
MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
endpoint.setMethod(methodToUse);
processListener(endpoint, rabbitListener, bean, methodToUse, beanName);
return processListener(endpoint, rabbitListener, bean, methodToUse, beanName);
}
private Method checkProxy(Method methodArg, Object bean) {
@@ -401,13 +403,14 @@ public class RabbitListenerAnnotationBeanPostProcessor
return method;
}
protected void processListener(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object bean,
Object target, String beanName) {
protected Collection<Declarable> processListener(MethodRabbitListenerEndpoint endpoint,
RabbitListener rabbitListener, Object bean, Object target, String beanName) {
final List<Declarable> declarables = new ArrayList<>();
endpoint.setBean(bean);
endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
endpoint.setId(getEndpointId(rabbitListener));
endpoint.setQueueNames(resolveQueues(rabbitListener));
endpoint.setQueueNames(resolveQueues(rabbitListener, declarables));
endpoint.setConcurrency(resolveExpressionAsStringOrInteger(rabbitListener.concurrency(), "concurrency"));
endpoint.setBeanFactory(this.beanFactory);
endpoint.setReturnExceptions(resolveExpressionAsBoolean(rabbitListener.returnExceptions()));
@@ -456,6 +459,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
RabbitListenerContainerFactory<?> factory = resolveContainerFactory(rabbitListener, target, beanName);
this.registrar.registerEndpoint(endpoint, factory);
return declarables;
}
private void resolveAckMode(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener) {
@@ -562,7 +566,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
}
private String[] resolveQueues(RabbitListener rabbitListener) {
private String[] resolveQueues(RabbitListener rabbitListener, Collection<Declarable> declarables) {
String[] queues = rabbitListener.queues();
QueueBinding[] bindings = rabbitListener.bindings();
org.springframework.amqp.rabbit.annotation.Queue[] queuesToDeclare = rabbitListener.queuesToDeclare();
@@ -578,7 +582,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
"@RabbitListener can have only one of 'queues', 'queuesToDeclare', or 'bindings'");
}
for (int i = 0; i < queuesToDeclare.length; i++) {
result.add(declareQueue(queuesToDeclare[i]));
result.add(declareQueue(queuesToDeclare[i], declarables));
}
}
if (bindings.length > 0) {
@@ -586,7 +590,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
throw new BeanInitializationException(
"@RabbitListener can have only one of 'queues', 'queuesToDeclare', or 'bindings'");
}
return registerBeansForDeclaration(rabbitListener);
return registerBeansForDeclaration(rabbitListener, declarables);
}
return result.toArray(new String[result.size()]);
}
@@ -618,19 +622,20 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
}
private String[] registerBeansForDeclaration(RabbitListener rabbitListener) {
private String[] registerBeansForDeclaration(RabbitListener rabbitListener, Collection<Declarable> declarables) {
List<String> queues = new ArrayList<String>();
if (this.beanFactory instanceof ConfigurableBeanFactory) {
for (QueueBinding binding : rabbitListener.bindings()) {
String queueName = declareQueue(binding.value());
String queueName = declareQueue(binding.value(), declarables);
queues.add(queueName);
declareExchangeAndBinding(binding, queueName);
declareExchangeAndBinding(binding, queueName, declarables);
}
}
return queues.toArray(new String[queues.size()]);
}
private String declareQueue(org.springframework.amqp.rabbit.annotation.Queue bindingQueue) {
private String declareQueue(org.springframework.amqp.rabbit.annotation.Queue bindingQueue,
Collection<Declarable> declarables) {
String queueName = (String) resolveExpression(bindingQueue.value());
boolean isAnonymous = false;
if (!StringUtils.hasText(queueName)) {
@@ -649,10 +654,11 @@ public class RabbitListenerAnnotationBeanPostProcessor
queue.setAdminsThatShouldDeclare((Object[]) bindingQueue.admins());
}
queue.setShouldDeclare(resolveExpressionAsBoolean(bindingQueue.declare()));
declarables.add(queue);
return queueName;
}
private void declareExchangeAndBinding(QueueBinding binding, String queueName) {
private void declareExchangeAndBinding(QueueBinding binding, String queueName, Collection<Declarable> declarables) {
org.springframework.amqp.rabbit.annotation.Exchange bindingExchange = binding.exchange();
String exchangeName = resolveExpressionAsString(bindingExchange.value(), "@Exchange.exchange");
Assert.isTrue(StringUtils.hasText(exchangeName), () -> "Exchange name required; binding queue " + queueName);
@@ -697,10 +703,12 @@ public class RabbitListenerAnnotationBeanPostProcessor
((ConfigurableBeanFactory) this.beanFactory)
.registerSingleton(exchangeName + ++this.increment, exchange);
registerBindings(binding, queueName, exchangeName, exchangeType);
registerBindings(binding, queueName, exchangeName, exchangeType, declarables);
declarables.add(exchange);
}
private void registerBindings(QueueBinding binding, String queueName, String exchangeName, String exchangeType) {
private void registerBindings(QueueBinding binding, String queueName, String exchangeName, String exchangeType,
Collection<Declarable> declarables) {
final List<String> routingKeys;
if (exchangeType.equals(ExchangeTypes.FANOUT) || binding.key().length == 0) {
routingKeys = Collections.singletonList("");
@@ -725,6 +733,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
((ConfigurableBeanFactory) this.beanFactory)
.registerSingleton(exchangeName + "." + queueName + ++this.increment, actualBinding);
declarables.add(actualBinding);
}
}
@@ -815,7 +824,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
}
private String resolveExpressionAsString(String value, String attribute) {
protected String resolveExpressionAsString(String value, String attribute) {
Object resolved = resolveExpression(value);
if (resolved instanceof String) {
return (String) resolved;

116
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryContextWrapper.java Normal file
View File

@@ -0,0 +1,116 @@
/*
* Copyright 2020 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
*
* https://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.amqp.rabbit.connection;
import java.util.concurrent.Callable;
import org.springframework.util.StringUtils;
/**
* Helper class to handle {@link ConnectionFactory} context binding and unbinding when executing instructions.
*
* @author Wander Costa
*/
public class ConnectionFactoryContextWrapper {
private final ConnectionFactory connectionFactory;
public ConnectionFactoryContextWrapper(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
/**
* Executes a {@link Callable} binding to the default {@link ConnectionFactory} and finally unbinding it.
*
* @param callable the {@link Callable} object to be executed.
* @param <T> the return type.
* @return the result of the {@link Callable}.
* @throws Exception when an Exception is thrown by the {@link Callable}.
*/
public <T> T call(final Callable<T> callable) throws Exception {
return call(null, callable);
}
/**
* Executes a {@link Callable} binding the given {@link ConnectionFactory} and finally unbinding it.
*
* @param contextName the name of the context. In null, empty or blank, default context is bound.
* @param callable the {@link Callable} object to be executed.
* @param <T> the return type.
* @return the result of the {@link Callable}.
* @throws Exception when an Exception is thrown by the {@link Callable}.
*/
public <T> T call(final String contextName, final Callable<T> callable) throws Exception {
try {
bind(contextName);
return callable.call();
}
finally {
unbind(contextName);
}
}
/**
* Executes a {@link Runnable} binding to the default {@link ConnectionFactory} and finally unbinding it.
*
* @param runnable the {@link Runnable} object to be executed.
* @throws RuntimeException when a RuntimeException is thrown by the {@link Runnable}.
*/
public void run(final Runnable runnable) {
run(null, runnable);
}
/**
* Executes a {@link Runnable} binding the given {@link ConnectionFactory} and finally unbinding it.
*
* @param contextName the name of the context. In null, empty or blank, default context is bound.
* @param runnable the {@link Runnable} object to be executed.
* @throws RuntimeException when a RuntimeException is thrown by the {@link Runnable}.
*/
public void run(final String contextName, final Runnable runnable) {
try {
bind(contextName);
runnable.run();
}
finally {
unbind(contextName);
}
}
/**
* Binds the context.
*
* @param contextName the name of the context for the connection factory.
*/
private void bind(final String contextName) {
if (StringUtils.hasText(contextName)) {
SimpleResourceHolder.bind(this.connectionFactory, contextName);
}
}
/**
* Unbinds the context.
*
* @param contextName the name of the context for the connection factory.
*/
private void unbind(final String contextName) {
if (StringUtils.hasText(contextName)) {
SimpleResourceHolder.unbind(this.connectionFactory);
}
}
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.

332
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MockMultiRabbitTests.java Normal file
View File

@@ -0,0 +1,332 @@
/*
* Copyright 2020 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
*
* https://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.amqp.rabbit.annotation;
import java.util.HashMap;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.amqp.core.AbstractExchange;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.rabbit.config.MessageListenerTestContainer;
import org.springframework.amqp.rabbit.config.RabbitListenerContainerTestFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
import org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
/**
* @author Wander Costa
*/
class MockMultiRabbitTests {
@Test
@DisplayName("Test instantiation of multiple message listeners")
void multipleSimpleMessageListeners() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(MultiConfig.class,
SimpleMessageListenerTestBean.class);
Map<String, RabbitListenerContainerTestFactory> factories = context
.getBeansOfType(RabbitListenerContainerTestFactory.class, false, false);
Assertions.assertThat(factories).hasSize(3);
factories.values().forEach(factory -> {
Assertions.assertThat(factory.getListenerContainers().size())
.as("One container should have been registered").isEqualTo(1);
MessageListenerTestContainer container = factory.getListenerContainers().get(0);
RabbitListenerEndpoint endpoint = container.getEndpoint();
Assertions.assertThat(endpoint.getClass()).as("Wrong endpoint type")
.isEqualTo(MethodRabbitListenerEndpoint.class);
MethodRabbitListenerEndpoint methodEndpoint = (MethodRabbitListenerEndpoint) endpoint;
Assertions.assertThat(methodEndpoint.getBean()).isNotNull();
Assertions.assertThat(methodEndpoint.getMethod()).isNotNull();
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
methodEndpoint.setupListenerContainer(listenerContainer);
Assertions.assertThat(listenerContainer.getMessageListener()).isNotNull();
});
context.close(); // Close and stop the listeners
}
@Test
@DisplayName("Test declarables matching the proper declaring admin")
void testDeclarablesMatchProperRabbitAdmin() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(MultiConfig.class,
AutoBindingListenerTestBeans.class);
Map<String, RabbitListenerContainerTestFactory> factories = context
.getBeansOfType(RabbitListenerContainerTestFactory.class, false, false);
Assertions.assertThat(factories).hasSize(3);
BiFunction<RabbitAdmin, Declarable, Boolean> declares = (admin, dec) -> dec.getDeclaringAdmins().size() == 1
&& dec.getDeclaringAdmins().contains(admin.getBeanName());
Map<String, AbstractExchange> exchanges = context.getBeansOfType(AbstractExchange.class, false, false)
.values().stream().collect(Collectors.toMap(AbstractExchange::getName, v -> v));
Assertions.assertThat(exchanges).hasSize(3);
Assertions.assertThat(declares.apply(MultiConfig.DEFAULT_RABBIT_ADMIN, exchanges.get("testExchange"))).isTrue();
Assertions.assertThat(declares.apply(MultiConfig.RABBIT_ADMIN_BROKER_B, exchanges.get("testExchangeB")))
.isTrue();
Assertions.assertThat(declares.apply(MultiConfig.RABBIT_ADMIN_BROKER_C, exchanges.get("testExchangeC")))
.isTrue();
Map<String, org.springframework.amqp.core.Queue> queues = context
.getBeansOfType(org.springframework.amqp.core.Queue.class, false, false)
.values().stream().collect(Collectors.toMap(org.springframework.amqp.core.Queue::getName, v -> v));
Assertions.assertThat(queues).hasSize(3);
Assertions.assertThat(declares.apply(MultiConfig.DEFAULT_RABBIT_ADMIN, queues.get("testQueue"))).isTrue();
Assertions.assertThat(declares.apply(MultiConfig.RABBIT_ADMIN_BROKER_B, queues.get("testQueueB"))).isTrue();
Assertions.assertThat(declares.apply(MultiConfig.RABBIT_ADMIN_BROKER_C, queues.get("testQueueC"))).isTrue();
Map<String, Binding> bindings = context.getBeansOfType(Binding.class, false, false)
.values().stream().collect(Collectors.toMap(Binding::getRoutingKey, v -> v));
Assertions.assertThat(bindings).hasSize(3);
Assertions.assertThat(declares.apply(MultiConfig.DEFAULT_RABBIT_ADMIN, bindings.get("testKey"))).isTrue();
Assertions.assertThat(declares.apply(MultiConfig.RABBIT_ADMIN_BROKER_B, bindings.get("testKeyB"))).isTrue();
Assertions.assertThat(declares.apply(MultiConfig.RABBIT_ADMIN_BROKER_C, bindings.get("testKeyC"))).isTrue();
context.close(); // Close and stop the listeners
}
@Test
@DisplayName("Test stand-alone declarable not associated to any declaring admin")
void testStandAloneDeclarablesNotEnhancedWithSpecificDeclaringAdmin() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(MultiConfig.class,
StandAloneDeclarablesConfig.class, AutoBindingListenerTestBeans.class);
Declarable exchange = context.getBean(StandAloneDeclarablesConfig.EXCHANGE, AbstractExchange.class);
Assertions.assertThat(exchange.getDeclaringAdmins()).isEmpty();
Declarable queue = context.getBean(StandAloneDeclarablesConfig.QUEUE,
org.springframework.amqp.core.Queue.class);
Assertions.assertThat(queue.getDeclaringAdmins()).isEmpty();
Declarable binding = context.getBean(StandAloneDeclarablesConfig.BINDING, Binding.class);
Assertions.assertThat(binding.getDeclaringAdmins()).isEmpty();
context.close(); // Close and stop the listeners
}
@Test
@DisplayName("Test creation of connections at the proper brokers")
void testCreationOfConnections() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(MultiConfig.class,
AutoBindingListenerTestBeans.class);
final RabbitTemplate rabbitTemplate = new RabbitTemplate(MultiConfig.ROUTING_CONNECTION_FACTORY);
Mockito.verify(MultiConfig.DEFAULT_CONNECTION_FACTORY, Mockito.never()).createConnection();
Mockito.verify(MultiConfig.DEFAULT_CONNECTION, Mockito.never()).createChannel(false);
rabbitTemplate.convertAndSend("messageDefaultBroker");
Mockito.verify(MultiConfig.DEFAULT_CONNECTION_FACTORY).createConnection();
Mockito.verify(MultiConfig.DEFAULT_CONNECTION).createChannel(false);
Mockito.verify(MultiConfig.CONNECTION_FACTORY_BROKER_B, Mockito.never()).createConnection();
Mockito.verify(MultiConfig.CONNECTION_BROKER_B, Mockito.never()).createChannel(false);
SimpleResourceHolder.bind(MultiConfig.ROUTING_CONNECTION_FACTORY, "brokerB");
rabbitTemplate.convertAndSend("messageToBrokerB");
SimpleResourceHolder.unbind(MultiConfig.ROUTING_CONNECTION_FACTORY);
Mockito.verify(MultiConfig.CONNECTION_FACTORY_BROKER_B).createConnection();
Mockito.verify(MultiConfig.CONNECTION_BROKER_B).createChannel(false);
Mockito.verify(MultiConfig.CONNECTION_FACTORY_BROKER_C, Mockito.never()).createConnection();
Mockito.verify(MultiConfig.CONNECTION_BROKER_C, Mockito.never()).createChannel(false);
SimpleResourceHolder.bind(MultiConfig.ROUTING_CONNECTION_FACTORY, "brokerC");
rabbitTemplate.convertAndSend("messageToBrokerC");
SimpleResourceHolder.unbind(MultiConfig.ROUTING_CONNECTION_FACTORY);
Mockito.verify(MultiConfig.CONNECTION_FACTORY_BROKER_C).createConnection();
Mockito.verify(MultiConfig.CONNECTION_BROKER_C).createChannel(false);
context.close(); // Close and stop the listeners
}
@Component
static class AutoBindingListenerTestBeans {
@RabbitListener(bindings = @QueueBinding(
exchange = @Exchange("testExchange"),
value = @Queue("testQueue"),
key = "testKey"))
public void handleIt(String body) {
}
@RabbitListener(containerFactory = "brokerB", bindings = @QueueBinding(
exchange = @Exchange("testExchangeB"),
value = @Queue("testQueueB"),
key = "testKeyB"))
public void handleItB(String body) {
}
@RabbitListener(containerFactory = "brokerC", bindings = @QueueBinding(
exchange = @Exchange("testExchangeC"),
value = @Queue("testQueueC"),
key = "testKeyC"))
public void handleItC(String body) {
}
}
@Component
static class SimpleMessageListenerTestBean {
@RabbitListener(queues = "testQueue")
public void handleIt(String body) {
}
@RabbitListener(queues = "testQueueB", containerFactory = "brokerB")
public void handleItB(String body) {
}
@RabbitListener(queues = "testQueueC", containerFactory = "brokerC")
public void handleItC(String body) {
}
}
@Configuration
@PropertySource("classpath:/org/springframework/amqp/rabbit/annotation/queue-annotation.properties")
static class MultiConfig {
static final SimpleRoutingConnectionFactory ROUTING_CONNECTION_FACTORY = new SimpleRoutingConnectionFactory();
static final ConnectionFactory DEFAULT_CONNECTION_FACTORY = Mockito.mock(ConnectionFactory.class);
static final ConnectionFactory CONNECTION_FACTORY_BROKER_B = Mockito.mock(ConnectionFactory.class);
static final ConnectionFactory CONNECTION_FACTORY_BROKER_C = Mockito.mock(ConnectionFactory.class);
static final Connection DEFAULT_CONNECTION = Mockito.mock(Connection.class);
static final Connection CONNECTION_BROKER_B = Mockito.mock(Connection.class);
static final Connection CONNECTION_BROKER_C = Mockito.mock(Connection.class);
static final Channel DEFAULT_CHANNEL = Mockito.mock(Channel.class);
static final Channel CHANNEL_BROKER_B = Mockito.mock(Channel.class);
static final Channel CHANNEL_BROKER_C = Mockito.mock(Channel.class);
static {
final Map<Object, ConnectionFactory> targetConnectionFactories = new HashMap<>();
targetConnectionFactories.put("brokerB", CONNECTION_FACTORY_BROKER_B);
targetConnectionFactories.put("brokerC", CONNECTION_FACTORY_BROKER_C);
ROUTING_CONNECTION_FACTORY.setDefaultTargetConnectionFactory(DEFAULT_CONNECTION_FACTORY);
ROUTING_CONNECTION_FACTORY.setTargetConnectionFactories(targetConnectionFactories);
Mockito.when(DEFAULT_CONNECTION_FACTORY.createConnection()).thenReturn(DEFAULT_CONNECTION);
Mockito.when(CONNECTION_FACTORY_BROKER_B.createConnection()).thenReturn(CONNECTION_BROKER_B);
Mockito.when(CONNECTION_FACTORY_BROKER_C.createConnection()).thenReturn(CONNECTION_BROKER_C);
Mockito.when(DEFAULT_CONNECTION.createChannel(false)).thenReturn(DEFAULT_CHANNEL);
Mockito.when(CONNECTION_BROKER_B.createChannel(false)).thenReturn(CHANNEL_BROKER_B);
Mockito.when(CONNECTION_BROKER_C.createChannel(false)).thenReturn(CHANNEL_BROKER_C);
}
static final RabbitAdmin DEFAULT_RABBIT_ADMIN = new RabbitAdmin(DEFAULT_CONNECTION_FACTORY);
static final RabbitAdmin RABBIT_ADMIN_BROKER_B = new RabbitAdmin(CONNECTION_FACTORY_BROKER_B);
static final RabbitAdmin RABBIT_ADMIN_BROKER_C = new RabbitAdmin(CONNECTION_FACTORY_BROKER_C);
@Bean
public RabbitListenerAnnotationBeanPostProcessor postProcessor() {
MultiRabbitListenerAnnotationBeanPostProcessor postProcessor
= new MultiRabbitListenerAnnotationBeanPostProcessor();
postProcessor.setEndpointRegistry(rabbitListenerEndpointRegistry());
postProcessor.setContainerFactoryBeanName("defaultContainerFactory");
return postProcessor;
}
@Bean("defaultRabbitAdmin")
public RabbitAdmin defaultRabbitAdmin() {
return DEFAULT_RABBIT_ADMIN;
}
@Bean("brokerB-admin")
public RabbitAdmin rabbitAdminBrokerB() {
return RABBIT_ADMIN_BROKER_B;
}
@Bean("brokerC-admin")
public RabbitAdmin rabbitAdminBrokerC() {
return RABBIT_ADMIN_BROKER_C;
}
@Bean("defaultContainerFactory")
public RabbitListenerContainerTestFactory defaultContainerFactory() {
return new RabbitListenerContainerTestFactory();
}
@Bean("brokerB")
public RabbitListenerContainerTestFactory containerFactoryBrokerB() {
return new RabbitListenerContainerTestFactory();
}
@Bean("brokerC")
public RabbitListenerContainerTestFactory containerFactoryBrokerC() {
return new RabbitListenerContainerTestFactory();
}
@Bean
public RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry() {
return new RabbitListenerEndpointRegistry();
}
@Bean
public static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() {
return new PropertySourcesPlaceholderConfigurer();
}
}
@Configuration
static class StandAloneDeclarablesConfig {
static final String EXCHANGE = "standAloneExchange";
static final String QUEUE = "standAloneQueue";
static final String BINDING = "standAloneBinding";
@Bean(name = EXCHANGE)
public AbstractExchange exchange() {
return new DirectExchange(EXCHANGE);
}
@Bean(name = QUEUE)
public org.springframework.amqp.core.Queue queue() {
return new org.springframework.amqp.core.Queue(QUEUE);
}
@Bean(name = BINDING)
public Binding binding() {
return new Binding(QUEUE, Binding.DestinationType.QUEUE, EXCHANGE, BINDING, null);
}
}
}

59
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/MultiRabbitListenerAnnotationBeanPostProcessorCompatibilityTests.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* Copyright 2020 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
*
* https://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.amqp.rabbit.annotation;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
/**
* This test is an extension of {@link RabbitListenerAnnotationBeanPostProcessorTests}
* in order to guarantee the compatibility of MultiRabbit with previous use cases. This
* ensures that multirabbit does not break single rabbit applications.
*
* @author Wander Costa
*/
class MultiRabbitListenerAnnotationBeanPostProcessorCompatibilityTests
extends RabbitListenerAnnotationBeanPostProcessorTests {
@Override
protected Class<?> getConfigClass() {
return MultiConfig.class;
}
@Configuration
@PropertySource("classpath:/org/springframework/amqp/rabbit/annotation/queue-annotation.properties")
static class MultiConfig extends RabbitListenerAnnotationBeanPostProcessorTests.Config {
@Bean
@Override
public MultiRabbitListenerAnnotationBeanPostProcessor postProcessor() {
MultiRabbitListenerAnnotationBeanPostProcessor postProcessor
= new MultiRabbitListenerAnnotationBeanPostProcessor();
postProcessor.setEndpointRegistry(rabbitListenerEndpointRegistry());
postProcessor.setContainerFactoryBeanName("testFactory");
return postProcessor;
}
@Bean
public RabbitAdmin defaultRabbitAdmin() {
return new RabbitAdmin(new SingleConnectionFactory());
}
}
}

30
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java
View File

@@ -66,10 +66,14 @@ import org.springframework.stereotype.Component;
*/
public class RabbitListenerAnnotationBeanPostProcessorTests {
protected Class<?> getConfigClass() {
return Config.class;
}
@Test
public void simpleMessageListener() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, SimpleMessageListenerTestBean.class);
getConfigClass(), SimpleMessageListenerTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("One container should have been registered").isEqualTo(1);
@@ -93,7 +97,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void simpleMessageListenerWithMixedAnnotations() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, SimpleMessageListenerWithMixedAnnotationsTestBean.class);
getConfigClass(), SimpleMessageListenerWithMixedAnnotationsTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("One container should have been registered").isEqualTo(1);
@@ -121,7 +125,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void metaAnnotationIsDiscovered() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, MetaAnnotationTestBean.class);
getConfigClass(), MetaAnnotationTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(2);
@@ -163,7 +167,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void multipleQueueNamesTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, MultipleQueueNamesTestBean.class);
getConfigClass(), MultipleQueueNamesTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(1);
@@ -178,7 +182,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void multipleQueuesTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, MultipleQueuesTestBean.class);
getConfigClass(), MultipleQueuesTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(1);
@@ -193,7 +197,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void mixedQueuesAndQueueNamesTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, MixedQueuesAndQueueNamesTestBean.class);
getConfigClass(), MixedQueuesAndQueueNamesTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(1);
@@ -209,7 +213,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void propertyResolvingToExpressionTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(
Config.class, PropertyResolvingToExpressionTestBean.class);
getConfigClass(), PropertyResolvingToExpressionTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(1);
@@ -224,7 +228,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void invalidValueInAnnotationTestBean() {
try {
new AnnotationConfigApplicationContext(Config.class, InvalidValueInAnnotationTestBean.class).close();
new AnnotationConfigApplicationContext(getConfigClass(), InvalidValueInAnnotationTestBean.class).close();
}
catch (BeanCreationException e) {
assertThat(e.getCause()).isInstanceOf(IllegalArgumentException.class);
@@ -235,7 +239,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void multipleRoutingKeysTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(Config.class,
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(getConfigClass(),
MultipleRoutingKeysTestBean.class);
RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class);
@@ -265,8 +269,8 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
}
@Test
public void customExhangeTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(Config.class,
public void customExchangeTestBean() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(getConfigClass(),
CustomExchangeTestBean.class);
final Collection<CustomExchange> exchanges = context.getBeansOfType(CustomExchange.class).values();
@@ -280,7 +284,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
@Test
public void queuesToDeclare() {
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(Config.class,
ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(getConfigClass(),
QueuesToDeclareTestBean.class);
final List<Queue> queues = new ArrayList<>(context.getBeansOfType(Queue.class).values());
@@ -309,7 +313,7 @@ public class RabbitListenerAnnotationBeanPostProcessorTests {
final ExecutorService executorService = Executors.newFixedThreadPool(concurrencyLevel);
try {
for (int i = 0; i < 1000; ++i) {
final ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
final ConfigurableApplicationContext context = new AnnotationConfigApplicationContext(getConfigClass());
try {
final Callable<?> task = () -> context.getBeanFactory().createBean(BeanForConcurrencyTesting.class);
final List<? extends Future<?>> futures = executorService

75
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryContextWrapperTests.java Normal file
View File

@@ -0,0 +1,75 @@
/*
* Copyright 2020 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
*
* https://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.amqp.rabbit.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.util.concurrent.Callable;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
/**
* @author Wander Costa
*/
class ConnectionFactoryContextWrapperTests {
@Test
@DisplayName("Test execution of callable in context wrapper")
void testExecuteCallable() throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
ConnectionFactoryContextWrapper wrapper = new ConnectionFactoryContextWrapper(connectionFactory);
String dummyContext = "dummy-context";
String expectedResult = "dummy-result";
Callable<String> callable = () -> expectedResult;
String actualResult = wrapper.call(dummyContext, callable);
assertThat(expectedResult).isEqualTo(actualResult);
}
@Test
@DisplayName("Test exception of runnable in context wrapper")
void testExecuteRunnable() {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
ConnectionFactoryContextWrapper wrapper = new ConnectionFactoryContextWrapper(connectionFactory);
String dummyContext = "dummy-context";
Runnable runnable = mock(Runnable.class);
wrapper.run(dummyContext, runnable);
verify(runnable).run();
}
@Test
@DisplayName("Test exception interception with runnable in context wrapper")
void testExecuteCallableThrowsRuntimeException() {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
ConnectionFactoryContextWrapper wrapper = new ConnectionFactoryContextWrapper(connectionFactory);
String dummyContext = "dummy-context";
Callable<String> callable = () -> {
throw new RuntimeException("dummy-exception");
};
Assertions.assertThrows(RuntimeException.class, () -> wrapper.call(dummyContext, callable));
}
}