diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index b8dacc30..10ceb33f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -328,7 +328,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen } else { try { - channel.getTargetChannel().close(); // to remove it from auto-recovery if so configured + Channel target = channel.getTargetChannel(); + if (target != null) { + target.close(); // to remove it from auto-recovery if so configured + } } catch (AlreadyClosedException e) { if (logger.isTraceEnabled()) { @@ -592,7 +595,9 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen // Handle close method: don't pass the call on. if (active) { synchronized (this.channelList) { - if (!RabbitUtils.isPhysicalCloseRequired() && this.channelList.size() < getChannelCacheSize()) { + if (!RabbitUtils.isPhysicalCloseRequired() && + (this.channelList.size() < getChannelCacheSize() + || this.channelList.contains((ChannelProxy) proxy))) { logicalClose((ChannelProxy) proxy); // Remain open in the channel list. releasePermit(); @@ -656,9 +661,15 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen * @param proxy the channel to close */ private void logicalClose(ChannelProxy proxy) throws Exception { + if (target == null) { + return; + } if (this.target != null && !this.target.isOpen()) { synchronized (targetMonitor) { if (this.target != null && !this.target.isOpen()) { + if (this.channelList.contains(proxy)) { + this.channelList.remove(proxy); + } this.target = null; return; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index f81444fc..0c1d7e85 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -10,6 +10,7 @@ * 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.junit.Assert.assertEquals; @@ -1075,4 +1076,27 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest assertSame(mockChannel, proxy.getTargetChannel()); } + @Test + public void testChannelCloseIdempotency() throws IOException { + com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); + com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel); + when(mockChannel.isOpen()).thenReturn(true).thenReturn(false); + + CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); + Connection con = ccf.createConnection(); + + Channel channel = con.createChannel(false); +// RabbitUtils.setPhysicalCloseRequired(true); + channel.close(); // should be ignored, and placed into channel cache. + channel.close(); // physically closed, so remove from the cache. + channel.close(); // physically closed and removed from the cache before, so void "close". + Channel channel2 = con.createChannel(false); + assertNotSame(channel, channel2); + } + }