Merge pull request #87 from garyrussell/AMQP-294

* garyrussell-AMQP-294:
  AMQP-294 Remove Class and Package Tangles

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -21,6 +21,7 @@ import java.util.concurrent.ExecutorService;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.Assert;
@@ -160,7 +161,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 				return new SimpleConnection(this.rabbitConnectionFactory.newConnection(this.executorService));
 			}
 		} catch (IOException e) {
-			throw RabbitUtils.convertRabbitAccessException(e);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -15,9 +15,8 @@ package org.springframework.amqp.rabbit.connection;
 
 import java.io.IOException;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.springframework.amqp.AmqpIOException;
+import org.springframework.amqp.rabbit.support.ConsumerChannelRegistry;
 import org.springframework.transaction.support.ResourceHolderSynchronization;
 import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -38,46 +37,6 @@ import com.rabbitmq.client.Channel;
  * @author Gary Russell
  */
 public class ConnectionFactoryUtils {
-
-	private static final Log logger = LogFactory.getLog(ConnectionFactoryUtils.class);
-
-	private static final ThreadLocal<Channel> consumerChannel = new ThreadLocal<Channel>();
-
-	/**
-	 * If a listener container is configured to use a RabbitTransactionManager, the
-	 * consumer's channel is registered here so that it is used as the bound resource
-	 * when the transaction actually starts. It is normally not necessary to use
-	 * an external transaction manager because local transactions work the same in that
-	 * the channel is bound to the thread. This is for the case when a user happens
-	 * to wire in a RabbitTransactionManager.
-	 * @param channel
-	 */
-	public static void registerConsumerChannel(Channel channel) {
-		if (logger.isDebugEnabled()) {
-			logger.debug("Registering consumer channel" + channel);
-		}
-		consumerChannel.set(channel);
-	}
-
-	/**
-	 * See registerConsumerChannel. This method is called to unregister
-	 * the channel when the consumer exits.
-	 */
-	public static void unRegisterConsumerChannel() {
-		if (logger.isDebugEnabled()) {
-			logger.debug("Unregistering consumer channel" + consumerChannel.get());
-		}
-		consumerChannel.remove();
-	}
-
-	/**
-	 * See registerConsumerChannel. This method is called to retrieve the
-	 * channel for this consumer.
-	 */
-	public static Channel getConsumerChannel() {
-		return consumerChannel.get();
-	}
-
 	/**
 	 * Determine whether the given RabbitMQ Channel is transactional, that is, bound to the current thread by Spring's
 	 * transaction facilities.
@@ -163,7 +122,7 @@ public class ConnectionFactoryUtils {
 				connection = resourceFactory.createConnection();
 				resourceHolderToUse.addConnection(connection);
 			}
-			channel = consumerChannel.get();
+			channel = ConsumerChannelRegistry.getConsumerChannel();
 			if (channel == null) {
 				channel = resourceFactory.createChannel(connection);
 			}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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
@@ -17,6 +17,7 @@ import java.io.IOException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 
@@ -25,6 +26,7 @@ import com.rabbitmq.client.Channel;
 /**
  * @author Mark Fisher
  * @author Dave Syer
+ * @author Gary Russell
  */
 public abstract class RabbitAccessor implements InitializingBean {
 
@@ -103,7 +105,7 @@ public abstract class RabbitAccessor implements InitializingBean {
 	}
 
 	protected RuntimeException convertRabbitAccessException(Exception ex) {
-		return RabbitUtils.convertRabbitAccessException(ex);
+		return RabbitExceptionTranslator.convertRabbitAccessException(ex);
 	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.ConsumerChannelRegistry;
 import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
 import org.springframework.transaction.support.ResourceHolderSupport;
 import org.springframework.util.Assert;
@@ -58,7 +59,7 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
 
 	private final Map<Connection, List<Channel>> channelsPerConnection = new HashMap<Connection, List<Channel>>();
 
-	private MultiValueMap<Channel, Long> deliveryTags = new LinkedMultiValueMap<Channel, Long>();
+	private final MultiValueMap<Channel, Long> deliveryTags = new LinkedMultiValueMap<Channel, Long>();
 
 	private boolean transactional;
 
@@ -154,7 +155,7 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
 	public void closeAll() {
 		for (Channel channel : this.channels) {
 			try {
-				if (channel != ConnectionFactoryUtils.getConsumerChannel()) {
+				if (channel != ConsumerChannelRegistry.getConsumerChannel()) {
 					channel.close();
 				}
 				else {

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

@@ -14,21 +14,15 @@
 package org.springframework.amqp.rabbit.connection;
 
 import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.ConnectException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.springframework.amqp.AmqpConnectException;
-import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpIOException;
-import org.springframework.amqp.AmqpUnsupportedEncodingException;
-import org.springframework.amqp.UncategorizedAmqpException;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.util.Assert;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ShutdownSignalException;
 
 /**
  * @author Mark Fisher
@@ -97,27 +91,6 @@ public abstract class RabbitUtils {
 		}
 	}
 
-	public static RuntimeException convertRabbitAccessException(Throwable ex) {
-		Assert.notNull(ex, "Exception must not be null");
-		if (ex instanceof AmqpException) {
-			return (AmqpException) ex;
-		}
-		if (ex instanceof ShutdownSignalException) {
-			return new AmqpConnectException((ShutdownSignalException) ex);
-		}
-		if (ex instanceof ConnectException) {
-			return new AmqpConnectException((ConnectException) ex);
-		}
-		if (ex instanceof IOException) {
-			return new AmqpIOException((IOException) ex);
-		}
-		if (ex instanceof UnsupportedEncodingException) {
-			return new AmqpUnsupportedEncodingException(ex);
-		}
-		// fallback
-		return new UncategorizedAmqpException(ex);
-	}
-
 	public static void closeMessageConsumer(Channel channel, String consumerTag, boolean transactional) {
 		if (!channel.isOpen()) {
 			return;
@@ -136,7 +109,7 @@ public abstract class RabbitUtils {
 			 * Messages are going to be lost in general.
 			 */
 		} catch (Exception ex) {
-			throw convertRabbitAccessException(ex);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(ex);
 		}
 	}
 
@@ -149,7 +122,7 @@ public abstract class RabbitUtils {
 		try {
 			channel.txSelect();
 		} catch (IOException e) {
-			throw RabbitUtils.convertRabbitAccessException(e);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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
@@ -14,8 +14,17 @@ package org.springframework.amqp.rabbit.connection;
 
 import java.io.IOException;
 
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
+
 import com.rabbitmq.client.Channel;
 
+/**
+ * Simply a Connection.
+ * @Dave Syer
+ * @author Gary Russell
+ * @since 1.0
+ *
+ */
 public class SimpleConnection implements Connection {
 
 	private final com.rabbitmq.client.Connection delegate;
@@ -33,7 +42,7 @@ public class SimpleConnection implements Connection {
 			}
 			return channel;
 		} catch (IOException e) {
-			throw RabbitUtils.convertRabbitAccessException(e);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 		}
 	}
 
@@ -41,7 +50,7 @@ public class SimpleConnection implements Connection {
 		try {
 			delegate.close();
 		} catch (IOException e) {
-			throw RabbitUtils.convertRabbitAccessException(e);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 		}
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -33,6 +33,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.AMQP.BasicProperties;
@@ -82,7 +83,7 @@ public class BlockingQueueConsumer {
 
 	private final ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter;
 
-	private Set<Long> deliveryTags = new LinkedHashSet<Long>();
+	private final Set<Long> deliveryTags = new LinkedHashSet<Long>();
 
 	private final boolean defaultRequeuRejected;
 
@@ -239,7 +240,7 @@ public class BlockingQueueConsumer {
 				}
 			}
 		} catch (IOException e) {
-			throw RabbitUtils.convertRabbitAccessException(e);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 		}
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -30,6 +30,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
 import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
+import org.springframework.amqp.rabbit.support.ConsumerChannelRegistry;
 import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.aop.Pointcut;
@@ -92,7 +93,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private volatile Advice[] adviceChain = new Advice[0];
 
-	private ActiveObjectCounter<BlockingQueueConsumer> cancellationLock = new ActiveObjectCounter<BlockingQueueConsumer>();
+	private final ActiveObjectCounter<BlockingQueueConsumer> cancellationLock = new ActiveObjectCounter<BlockingQueueConsumer>();
 
 	private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
 
@@ -102,7 +103,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		void invokeListener(Channel channel, Message message) throws Exception;
 	}
 
-	private ContainerDelegate delegate = new ContainerDelegate() {
+	private final ContainerDelegate delegate = new ContainerDelegate() {
 		public void invokeListener(Channel channel, Message message) throws Exception {
 			SimpleMessageListenerContainer.super.invokeListener(channel, message);
 		}
@@ -536,7 +537,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 					 * Register the consumer's channel so it will be used by the transaction manager
 					 * if it's an instance of RabbitTransactionManager.
 					 */
-					ConnectionFactoryUtils.registerConsumerChannel(consumer.getChannel());
+					ConsumerChannelRegistry.registerConsumerChannel(consumer.getChannel());
 				}
 
 				// Always better to stop receiving as soon as possible if
@@ -576,7 +577,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			}
 			finally {
 				if (SimpleMessageListenerContainer.this.transactionManager != null) {
-					ConnectionFactoryUtils.unRegisterConsumerChannel();
+					ConsumerChannelRegistry.unRegisterConsumerChannel();
 				}
 			}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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
@@ -26,11 +26,11 @@ import org.springframework.amqp.core.Address;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -118,6 +118,7 @@ import com.rabbitmq.client.Channel;
  * @author Mark Pollack
  * @author Mark Fisher
  * @author Dave Syer
+ * @author Gary Russell
  *
  * @see #setDelegate
  * @see #setDefaultListenerMethod
@@ -574,7 +575,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
 			channel.basicPublish(replyTo.getExchangeName(), replyTo.getRoutingKey(), this.mandatoryPublish,
 					this.immediatePublish, this.messagePropertiesConverter.fromMessageProperties(message.getMessageProperties(), encoding), message.getBody());
 		} catch (Exception ex) {
-			throw RabbitUtils.convertRabbitAccessException(ex);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(ex);
 		}
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ConsumerChannelRegistry.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2013 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on 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.support;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import com.rabbitmq.client.Channel;
+
+/**
+ * Consumers register their primary channels with this class. This is used
+ * to ensure that, when using transactions, the resource holder doesn't
+ * close the primary channel being used by the Consumer.
+ * This was previously in ConnectionFactoryUtils, but it caused a class
+ * tangle with RabbitResourceHolder.
+ *
+ * @author Gary Russell
+ * @since 1.2
+ *
+ */
+public class ConsumerChannelRegistry {
+
+	private static final Log logger = LogFactory.getLog(ConsumerChannelRegistry.class);
+
+	private static final ThreadLocal<Channel> consumerChannel = new ThreadLocal<Channel>();
+
+	/**
+	 * If a listener container is configured to use a RabbitTransactionManager, the
+	 * consumer's channel is registered here so that it is used as the bound resource
+	 * when the transaction actually starts. It is normally not necessary to use
+	 * an external transaction manager because local transactions work the same in that
+	 * the channel is bound to the thread. This is for the case when a user happens
+	 * to wire in a RabbitTransactionManager.
+	 * @param channel
+	 */
+	public static void registerConsumerChannel(Channel channel) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Registering consumer channel" + channel);
+		}
+		consumerChannel.set(channel);
+	}
+
+	/**
+	 * See registerConsumerChannel. This method is called to unregister
+	 * the channel when the consumer exits.
+	 */
+	public static void unRegisterConsumerChannel() {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Unregistering consumer channel" + consumerChannel.get());
+		}
+		consumerChannel.remove();
+	}
+
+	/**
+	 * See registerConsumerChannel. This method is called to retrieve the
+	 * channel for this consumer.
+	 */
+	public static Channel getConsumerChannel() {
+		return consumerChannel.get();
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2013 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
@@ -24,7 +24,6 @@ import java.util.Map;
 import org.springframework.amqp.AmqpUnsupportedEncodingException;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.util.CollectionUtils;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
@@ -35,6 +34,7 @@ import com.rabbitmq.client.LongString;
  * Default implementation of the {@link MessagePropertiesConverter} strategy.
  *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 1.0
  */
 public class DefaultMessagePropertiesConverter implements MessagePropertiesConverter {
@@ -154,7 +154,7 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve
 				return longString.getStream();
 			}
 		} catch (Exception e) {
-			throw RabbitUtils.convertRabbitAccessException(e);
+			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 		}
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2002-2013 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on 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.support;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.ConnectException;
+
+import org.springframework.amqp.AmqpConnectException;
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.AmqpIOException;
+import org.springframework.amqp.AmqpUnsupportedEncodingException;
+import org.springframework.amqp.UncategorizedAmqpException;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.client.ShutdownSignalException;
+
+/**
+ * Translates Rabbit Exceptions to the {@link AmqpException} class
+ * hierarchy.
+ * This functionality was previously in RabbitUtils, but that
+ * caused a package tange.
+ * @author Gary Russell
+ * @since 1.2
+ *
+ */
+public class RabbitExceptionTranslator {
+
+	public static RuntimeException convertRabbitAccessException(Throwable ex) {
+		Assert.notNull(ex, "Exception must not be null");
+		if (ex instanceof AmqpException) {
+			return (AmqpException) ex;
+		}
+		if (ex instanceof ShutdownSignalException) {
+			return new AmqpConnectException((ShutdownSignalException) ex);
+		}
+		if (ex instanceof ConnectException) {
+			return new AmqpConnectException((ConnectException) ex);
+		}
+		if (ex instanceof IOException) {
+			return new AmqpIOException((IOException) ex);
+		}
+		if (ex instanceof UnsupportedEncodingException) {
+			return new AmqpUnsupportedEncodingException(ex);
+		}
+		// fallback
+		return new UncategorizedAmqpException(ex);
+	}
+
+}