AMQP-456: Don't ERROR Log For Passive Declare

JIRA: https://jira.spring.io/browse/AMQP-456

Previously all abnormal channel closes were logged at ERROR level,
even NOT FOUND for passive declarations.

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java

@@ -20,6 +20,9 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Message;
@@ -38,6 +41,8 @@ import org.springframework.util.FileCopyUtils;
  */
 public abstract class AbstractCompressingPostProcessor implements MessagePostProcessor, Ordered {
 
+	private final Log logger = LogFactory.getLog(this.getClass());
+
 	private final boolean autoDecompress;
 
 	private int order;
@@ -75,7 +80,11 @@ public abstract class AbstractCompressingPostProcessor implements MessagePostPro
 			if (this.autoDecompress) {
 				messageProperties.setHeader(MessageProperties.SPRING_AUTO_DECOMPRESS, true);
 			}
-			return new Message(zipped.toByteArray(), messageProperties);
+			byte[] compressed = zipped.toByteArray();
+			if (logger.isTraceEnabled()) {
+				logger.trace("Compressed " + message.getBody().length + " to " + compressed.length);
+			}
+			return new Message(compressed, messageProperties);
 		}
 		catch (IOException e) {
 			throw new AmqpIOException(e);

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java

@@ -245,7 +245,12 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 
 	@Override
 	public void shutdownCompleted(ShutdownSignalException cause) {
-		if (!RabbitUtils.isNormalChannelClose(cause)) {
+		if (RabbitUtils.isPassiveDeclarationChannelClose(cause)) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Channel shutdown: " + cause.getMessage());
+			}
+		}
+		else if (!RabbitUtils.isNormalChannelClose(cause)) {
 			logger.error("Channel shutdown: " + cause.getMessage());
 		}
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java

@@ -191,6 +191,15 @@ public abstract class RabbitUtils {
 				&& "OK".equals(((AMQP.Channel.Close) shutdownReason).getReplyText());
 	}
 
+	public static boolean isPassiveDeclarationChannelClose(ShutdownSignalException sig) {
+		Object shutdownReason = determineShutdownReason(sig);
+		return shutdownReason instanceof AMQP.Channel.Close
+				&& AMQP.NOT_FOUND == ((AMQP.Channel.Close) shutdownReason).getReplyCode()
+				&& ((((AMQP.Channel.Close) shutdownReason).getClassId() == 40 // exchange
+					|| ((AMQP.Channel.Close) shutdownReason).getClassId() == 50) // queue
+					&& ((AMQP.Channel.Close) shutdownReason).getMethodId() == 10); // declare
+	}
+
 	protected static Object determineShutdownReason(ShutdownSignalException sig) {
 		if (shutDownSignalReasonMethod == null) {
 			return false;

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java

@@ -14,6 +14,7 @@
 package org.springframework.amqp.rabbit.core;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -22,9 +23,14 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.lang.reflect.Field;
 import java.util.HashMap;
@@ -45,11 +51,13 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Address;
 import org.springframework.amqp.core.AddressUtils;
 import org.springframework.amqp.core.Message;
@@ -71,6 +79,8 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.support.postprocessor.GUnzipPostProcessor;
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
 import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.amqp.utils.test.TestUtils;
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -88,6 +98,7 @@ import org.springframework.util.ReflectionUtils.FieldFilter;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.ShutdownSignalException;
 
 /**
  * @author Dave Syer
@@ -1140,6 +1151,64 @@ public class RabbitTemplateIntegrationTests {
 		}
 	}
 
+	@Test
+	public void testDegugLogOnPassiveDeclaration() {
+		CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
+		Log logger = spy(TestUtils.getPropertyValue(connectionFactory, "logger", Log.class));
+		when(logger.isDebugEnabled()).thenReturn(true);
+		new DirectFieldAccessor(connectionFactory).setPropertyValue("logger", logger);
+		RabbitTemplate template = new RabbitTemplate(connectionFactory);
+		final String queueName = UUID.randomUUID().toString();
+		final String exchangeName = UUID.randomUUID().toString();
+		try {
+			template.execute(new ChannelCallback<Void>() {
+
+				@Override
+				public Void doInRabbit(Channel channel) throws Exception {
+					channel.queueDeclarePassive(queueName);
+					return null;
+				}
+			});
+			fail("Expected exception");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(AmqpIOException.class));
+			assertThat(e.getCause(), instanceOf(IOException.class));
+			assertThat(e.getCause().getCause(), instanceOf(ShutdownSignalException.class));
+			assertThat(e.getCause().getCause().getMessage(), containsString("404"));
+		}
+		try {
+			template.execute(new ChannelCallback<Void>() {
+
+				@Override
+				public Void doInRabbit(Channel channel) throws Exception {
+					channel.exchangeDeclarePassive(exchangeName);
+					return null;
+				}
+			});
+			fail("Expected exception");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(AmqpIOException.class));
+			assertThat(e.getCause(), instanceOf(IOException.class));
+			assertThat(e.getCause().getCause(), instanceOf(ShutdownSignalException.class));
+			assertThat(e.getCause().getCause().getMessage(), containsString("404"));
+		}
+		verify(logger, never()).error(org.mockito.Matchers.any());
+		ArgumentCaptor<Object> logs = ArgumentCaptor.forClass(Object.class);
+		verify(logger, atLeast(2)).debug(logs.capture());
+		boolean queue = false;
+		boolean exchange = false;
+		for (Object log : logs.getAllValues()) {
+			String logMessage = (String) log;
+			queue |= (logMessage.contains(queueName) && logMessage.contains("404"));
+			exchange |= (logMessage.contains(queueName) && logMessage.contains("404"));
+		}
+		assertTrue(queue);
+		assertTrue(exchange);
+		connectionFactory.destroy();
+	}
+
 	@SuppressWarnings("serial")
 	private class PlannedException extends RuntimeException {
 		public PlannedException() {