From 41ccf8a1855c2435dc0171227956fd1aac18702e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 3 Apr 2015 11:51:31 +0300 Subject: [PATCH] AMQP-487: Fix NPE in the CachingConnectionFactory JIRA: https://jira.spring.io/browse/AMQP-487 The `ChannelProxy.target` can be closed externally in between several `close` operations on the proxy. The second `close()` caused an NPE before this fix. In addition the closed `ChannelProxy` may be returned from the cache. Prevent NPE on the `target.close()` operation. **Cherry-pick to 1.4.x** Conflicts: spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java Resolved. --- .../connection/CachingConnectionFactory.java | 15 +++++++++-- .../CachingConnectionFactoryTests.java | 26 ++++++++++++++++++- 2 files changed, 38 insertions(+), 3 deletions(-) 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); + } + }