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

AMQP-796: Fix Admin Transaction

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-796

If an admin uses a transactional `RabbitTemplate` it will start a transaction.
If the connection was opened due to a `RabbitTemplate` operation it should participate
in the same transaction.
Previously, the template used a second channel and treated it as a local transaction.

Also fix the `RabbitAdmin` so it does no work if there is nothing to declare.

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java

* Remove `RabbitTemplateTests` changes since they are not related to
the current state of the `RabbitAdmin`: the `RabbitTemplate`-based
constructor has been introduced since version `2.0`
This commit is contained in:
Gary Russell
2018-01-23 14:56:06 -05:00
committed by Artem Bilan
parent f78d916719
commit f786c5b59d
5 changed files with 67 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -138,6 +138,18 @@ public final class ConnectionFactoryUtils {
channel = ConsumerChannelRegistry.getConsumerChannel(connectionFactory);
if (channel == null && connection == null) {
connection = resourceFactory.createConnection();
if (resourceHolder == null) {
/*
* While creating a connection, a connection listener might have created a
* transactional channel and bound it to the transaction.
*/
resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager
.getResource(connectionFactory);
if (resourceHolder != null) {
channel = resourceHolder.getChannel();
resourceHolderToUse = resourceHolder;
}
}
resourceHolderToUse.addConnection(connection);
}
if (channel == null) {

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -497,6 +497,10 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
}
}
if (exchanges.size() == 0 && queues.size() == 0 && bindings.size() == 0) {
this.logger.debug("Nothing to declare");
return;
}
this.rabbitTemplate.execute(new ChannelCallback<Object>() {
@Override
public Object doInRabbit(Channel channel) throws Exception {

37
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java
View File

@@ -21,9 +21,12 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyMap;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.isNull;
import static org.mockito.Mockito.doAnswer;
@@ -55,6 +58,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.Cache
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -115,11 +119,14 @@ public class RabbitAdminDeclarationTests {
final List<Channel> mockChannels = new ArrayList<Channel>();
doAnswer(new Answer<com.rabbitmq.client.Connection>() {
private int connectionNumber;
@Override
public com.rabbitmq.client.Connection answer(InvocationOnMock invocation) throws Throwable {
com.rabbitmq.client.Connection connection = mock(com.rabbitmq.client.Connection.class);
doAnswer(new Answer<Channel>() {
private int channelNumber;
@Override
@@ -154,7 +161,7 @@ public class RabbitAdminDeclarationTests {
ccf.createConnection().close();
ccf.destroy();
assertEquals("Admin should not have created a channel", 0, mockChannels.size());
assertEquals("Admin should not have created a channel", 0, mockChannels.size());
}
@Test
@@ -234,7 +241,7 @@ public class RabbitAdminDeclarationTests {
verify(channel, never()).queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), any(Map.class));
}
@@ -275,7 +282,7 @@ public class RabbitAdminDeclarationTests {
verify(channel, never()).queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), any(Map.class));
}
@@ -293,7 +300,7 @@ public class RabbitAdminDeclarationTests {
.queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), isNull(Map.class));
verify(Config.channel2, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(),
anyBoolean(), anyMap());
anyBoolean(), anyMap());
verify(Config.channel2, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), anyMap());
context.close();
}
@@ -308,7 +315,7 @@ public class RabbitAdminDeclarationTests {
assertEquals(2, queue.getDeclaringAdmins().size());
queue.setAdminsThatShouldDeclare(admin1);
assertEquals(1, queue.getDeclaringAdmins().size());
queue.setAdminsThatShouldDeclare(new Object[] {null});
queue.setAdminsThatShouldDeclare(new Object[] { null });
assertEquals(0, queue.getDeclaringAdmins().size());
queue.setAdminsThatShouldDeclare(admin1, admin2);
assertEquals(2, queue.getDeclaringAdmins().size());
@@ -331,6 +338,26 @@ public class RabbitAdminDeclarationTests {
}
}
@Test
public void testNoOpWhenNothingToDeclare() throws Exception {
com.rabbitmq.client.ConnectionFactory cf = mock(com.rabbitmq.client.ConnectionFactory.class);
com.rabbitmq.client.Connection connection = mock(com.rabbitmq.client.Connection.class);
Channel channel = mock(Channel.class, "channel1");
given(channel.isOpen()).willReturn(true);
willReturn(connection).given(cf).newConnection(any(ExecutorService.class), anyString());
given(connection.isOpen()).willReturn(true);
given(connection.createChannel()).willReturn(channel);
CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
ccf.setExecutor(mock(ExecutorService.class));
RabbitTemplate rabbitTemplate = new RabbitTemplate(ccf);
RabbitAdmin admin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());
ApplicationContext ac = mock(ApplicationContext.class);
admin.setApplicationContext(ac);
admin.afterPropertiesSet();
ccf.createConnection();
verify(connection, never()).createChannel();
}
@Configuration
public static class Config {

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java
View File

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

21
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -268,6 +268,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Override
public void run() {
templateWithConfirmsEnabled.execute(new ChannelCallback<Object>() {
@Override
public Object doInRabbit(Channel channel) throws Exception {
try {
@@ -325,6 +326,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
final CountDownLatch latch = new CountDownLatch(1);
final List<Message> returns = new ArrayList<Message>();
templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode,
String replyText, String exchange, String routingKey) {
@@ -345,6 +347,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
final CountDownLatch latch = new CountDownLatch(1);
final List<Message> returns = new ArrayList<Message>();
templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode,
String replyText, String exchange, String routingKey) {
@@ -434,6 +437,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Override
public void run() {
template.execute(new ChannelCallback<Object>() {
@Override
public Object doInRabbit(Channel channel) throws Exception {
try {
@@ -443,9 +447,9 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
Thread.currentThread().interrupt();
}
template.doSend(channel, "", ROUTE,
new SimpleMessageConverter().toMessage("message", new MessageProperties()),
false,
new CorrelationData("def"));
new SimpleMessageConverter().toMessage("message", new MessageProperties()),
false,
new CorrelationData("def"));
threadSentLatch.countDown();
return null;
}
@@ -488,10 +492,12 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
} }).when(mockChannel).getNextPublishSeqNo();
}
}).when(mockChannel).getNextPublishSeqNo();
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setPublisherConfirms(true);
@@ -533,6 +539,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
@@ -580,6 +587,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
@@ -681,6 +689,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
}
});
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
@@ -689,6 +698,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
}
});
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
@@ -921,7 +931,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
try {
template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
}
catch (AmqpException e) { }
catch (AmqpException e) {
}
}
sentAll.countDown();
}