diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index 2a67df56..ab6bbde7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -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) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index 8e537120..95545e69 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -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() { @Override public Object doInRabbit(Channel channel) throws Exception { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java index cd8b78cd..cab28c88 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java @@ -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 mockChannels = new ArrayList(); doAnswer(new Answer() { + 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() { + 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 { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index 811109eb..45a6455a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java @@ -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. diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index cd87af01..e2ade55f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -268,6 +268,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { @Override public void run() { templateWithConfirmsEnabled.execute(new ChannelCallback() { + @Override public Object doInRabbit(Channel channel) throws Exception { try { @@ -325,6 +326,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { final CountDownLatch latch = new CountDownLatch(1); final List returns = new ArrayList(); 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 returns = new ArrayList(); 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() { + @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() { + @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() { + @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() { + @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(); }