Merge branch 'master' into gradle

pom.xml

@@ -50,6 +50,11 @@
 			</distributionManagement>
 		</profile>
 	</profiles>
+	<scm>
+		<url>http://git.springframework.org/spring-amqp</url>
+		<connection>scm:git:git://git.springsource.org/spring-amqp/spring-amqp.git</connection>
+		<developerConnection>scm:git:git://git.springsource.org/spring-amqp/spring-amqp.git</developerConnection>
+	</scm>
 	<distributionManagement>
 		<!-- see 'staging' profile for dry-run deployment settings -->
 		<downloadUrl>http://www.springsource.com/download/community</downloadUrl>

spring-amqp-core/src/main/java/org/springframework/amqp/support/converter/MessageConversionException.java

@@ -1,17 +1,14 @@
 /*
  * Copyright 2002-2010 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.
+ * 
+ * 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.support.converter;
@@ -19,10 +16,19 @@ package org.springframework.amqp.support.converter;
 import org.springframework.amqp.AmqpException;
 
 /**
+ * <p>
+ * Exception to be thrown by message converters if they encounter a problem with converting a message or object.
+ * </p>
+ * <p>
+ * N.B. this is <em>not</em> an {@link AmqpException} because it is a a client exception, not a protocol or broker
+ * problem.
+ * </p>
+ * 
  * @author Mark Fisher
+ * @author Dave Syer
  */
 @SuppressWarnings("serial")
-public class MessageConversionException extends AmqpException {
+public class MessageConversionException extends RuntimeException {
 
 	public MessageConversionException(String message, Throwable cause) {
 		super(message, cause);

spring-amqp-core/src/main/java/org/springframework/amqp/support/converter/SerializerMessageConverter.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 2002-2010 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.support.converter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.core.serializer.DefaultDeserializer;
+import org.springframework.core.serializer.DefaultSerializer;
+import org.springframework.core.serializer.Deserializer;
+import org.springframework.core.serializer.Serializer;
+
+/**
+ * Implementation of {@link MessageConverter} that can work with Strings or native objects of any kind via the
+ * {@link Serializer} and {@link Deserializer} abstractions in Spring. The {@link #toMessage(Object, MessageProperties)}
+ * method simply checks the type of the provided instance while the {@link #fromMessage(Message)} method relies upon the
+ * {@link MessageProperties#getContentType() content-type} of the provided Message.
+ * 
+ * @author Dave Syer
+ */
+public class SerializerMessageConverter implements MessageConverter {
+
+	public static final String DEFAULT_CHARSET = "UTF-8";
+
+	private volatile String defaultCharset = DEFAULT_CHARSET;
+
+	private volatile Serializer<Object> serializer = new DefaultSerializer();
+
+	private volatile Deserializer<Object> deserializer = new DefaultDeserializer();
+
+	private volatile boolean ignoreContentType = false;
+
+	/**
+	 * Flag to signal that the content type should be ignored and the deserializer used irrespective if it is a text
+	 * message. Defaults to false, in which case the default encoding is used to convert a text message to a String.
+	 * 
+	 * @param ignoreContentType the flag value to set
+	 */
+	public void setIgnoreContentType(boolean ignoreContentType) {
+		this.ignoreContentType = ignoreContentType;
+	}
+
+	/**
+	 * Specify the default charset to use when converting to or from text-based Message body content. If not specified,
+	 * the charset will be "UTF-8".
+	 */
+	public void setDefaultCharset(String defaultCharset) {
+		this.defaultCharset = (defaultCharset != null) ? defaultCharset : DEFAULT_CHARSET;
+	}
+
+	/**
+	 * The serializer to use for converting Java objects to message bodies.
+	 * 
+	 * @param serializer the serializer to set
+	 */
+	public void setSerializer(Serializer<Object> serializer) {
+		this.serializer = serializer;
+	}
+
+	/**
+	 * The deserializer to use for converting from message body to Java object.
+	 * 
+	 * @param deserializer the deserializer to set
+	 */
+	public void setDeserializer(Deserializer<Object> deserializer) {
+		this.deserializer = deserializer;
+	}
+
+	/**
+	 * Converts from a AMQP Message to an Object.
+	 */
+	public Object fromMessage(Message message) throws MessageConversionException {
+		Object content = null;
+		MessageProperties properties = message.getMessageProperties();
+		if (properties != null) {
+			String contentType = properties.getContentType();
+			if (contentType != null && contentType.startsWith("text") && !ignoreContentType) {
+				String encoding = properties.getContentEncoding();
+				if (encoding == null) {
+					encoding = this.defaultCharset;
+				}
+				try {
+					content = new String(message.getBody(), encoding);
+				} catch (UnsupportedEncodingException e) {
+					throw new MessageConversionException("failed to convert text-based Message content", e);
+				}
+			} else if (contentType != null && contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)
+					|| ignoreContentType) {
+				try {
+					content = deserializer.deserialize(new ByteArrayInputStream(message.getBody()));
+				} catch (IOException e) {
+					throw new MessageConversionException("Could not convert message body", e);
+				}
+			}
+		}
+		if (content == null) {
+			content = message.getBody();
+		}
+		return content;
+	}
+
+	/**
+	 * Creates an AMQP Message from the provided Object.
+	 */
+	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		byte[] bytes = null;
+		if (object instanceof String) {
+			try {
+				bytes = ((String) object).getBytes(this.defaultCharset);
+			} catch (UnsupportedEncodingException e) {
+				throw new MessageConversionException("failed to convert Message content", e);
+			}
+			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
+			messageProperties.setContentEncoding(this.defaultCharset);
+		} else if (object instanceof byte[]) {
+			bytes = (byte[]) object;
+			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_BYTES);
+		} else {
+			ByteArrayOutputStream output = new ByteArrayOutputStream();
+			try {
+				serializer.serialize(object, output);
+			} catch (IOException e) {
+				throw new MessageConversionException("Cannot convert object to bytes", e);
+			}
+			bytes = output.toByteArray();
+			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT);
+		}
+		if (bytes != null) {
+			messageProperties.setContentLength(bytes.length);
+		}
+		return new Message(bytes, messageProperties);
+	}
+
+}

spring-amqp-core/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java

@@ -16,12 +16,19 @@
 
 package org.springframework.amqp.support.converter;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
 import java.io.Serializable;
 import java.io.UnsupportedEncodingException;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;
+import org.springframework.util.ClassUtils;
 
 /**
  * Implementation of {@link MessageConverter} that can work with Strings, Serializable instances,
@@ -32,13 +39,32 @@ import org.springframework.amqp.utils.SerializationUtils;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  */
-public class SimpleMessageConverter implements MessageConverter {
+public class SimpleMessageConverter implements MessageConverter, BeanClassLoaderAware {
 
 	public static final String DEFAULT_CHARSET = "UTF-8";
 
-
 	private volatile String defaultCharset = DEFAULT_CHARSET;
 
+	private String codebaseUrl;
+
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+	
+	public void setBeanClassLoader(ClassLoader beanClassLoader) {
+		this.beanClassLoader = beanClassLoader;
+	}
+
+	/**
+	 * Set the codebase URL to download classes from if not found locally. Can consists of multiple URLs, separated by
+	 * spaces.
+	 * <p>
+	 * Follows RMI's codebase conventions for dynamic class download.
+	 * 
+	 * @see org.springframework.remoting.rmi.CodebaseAwareObjectInputStream
+	 * @see java.rmi.server.RMIClassLoader
+	 */
+	public void setCodebaseUrl(String codebaseUrl) {
+		this.codebaseUrl = codebaseUrl;
+	}
 
 	/**
 	 * Specify the default charset to use when converting to or from text-based
@@ -49,7 +75,7 @@ public class SimpleMessageConverter implements MessageConverter {
 	}
 
 	/**
-	 * Converts from a Rabbit Message to an Object.
+	 * Converts from a AMQP Message to an Object.
 	 */
 	public Object fromMessage(Message message) throws MessageConversionException {
 		Object content = null;
@@ -71,7 +97,15 @@ public class SimpleMessageConverter implements MessageConverter {
 			}
 			else if (contentType != null &&
 					contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
-				content = SerializationUtils.deserialize(message.getBody());
+				try {
+					content = SerializationUtils.deserialize(createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
+				} catch (IOException e) {
+					throw new MessageConversionException(
+							"failed to convert serialized Message content", e);
+				} catch (IllegalArgumentException e) {
+					throw new MessageConversionException(
+							"failed to convert serialized Message content", e);					
+				}
 			}
 		}
 		if (content == null) {
@@ -81,7 +115,7 @@ public class SimpleMessageConverter implements MessageConverter {
 	}
 
 	/**
-	 * Creates a Rabbit Mesasge from the provided Object.
+	 * Creates an AMQP Message from the provided Object.
 	 */
 	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
 		byte[] bytes = null;		
@@ -95,13 +129,18 @@ public class SimpleMessageConverter implements MessageConverter {
 			}
 			catch (UnsupportedEncodingException e) {
 				throw new MessageConversionException(
-						"failed to convert Message content", e);
+						"failed to convert to Message content", e);
 			}
 			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
 			messageProperties.setContentEncoding(this.defaultCharset);
 		}
 		else if (object instanceof Serializable) {
-			bytes = SerializationUtils.serialize(object);
+			try {
+				bytes = SerializationUtils.serialize(object);
+			} catch (IllegalArgumentException e) {
+				throw new MessageConversionException(
+						"failed to convert to serialized Message content", e);					
+			}
 			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT);
 		}
 		if (bytes != null) {
@@ -110,4 +149,17 @@ public class SimpleMessageConverter implements MessageConverter {
 		return new Message(bytes, messageProperties);
 	}
 
+	/**
+	 * Create an ObjectInputStream for the given InputStream and codebase. The default implementation creates a
+	 * CodebaseAwareObjectInputStream.
+	 * @param is the InputStream to read from
+	 * @param codebaseUrl the codebase URL to load classes from if not found locally (can be <code>null</code>)
+	 * @return the new ObjectInputStream instance to use
+	 * @throws IOException if creation of the ObjectInputStream failed
+	 * @see org.springframework.remoting.rmi.CodebaseAwareObjectInputStream
+	 */
+	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
+		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl);
+	}
+
 }

spring-amqp-core/src/main/java/org/springframework/amqp/utils/SerializationUtils.java

@@ -58,7 +58,22 @@ public class SerializationUtils {
 			return null;
 		}
 		try {
-			return new ObjectInputStream(new ByteArrayInputStream(bytes)).readObject();
+			return deserialize(new ObjectInputStream(new ByteArrayInputStream(bytes)));
+		} catch (IOException e) {
+			throw new IllegalArgumentException("Could not deserialize object", e);
+		}
+	}
+
+	/**
+	 * @param stream an object stream created from a serialized object
+	 * @return the result of deserializing the bytes
+	 */
+	public static Object deserialize(ObjectInputStream stream) {
+		if (stream == null) {
+			return null;
+		}
+		try {
+			return stream.readObject();
 		}
 		catch (IOException e) {
 			throw new IllegalArgumentException("Could not deserialize object", e);

spring-amqp-core/src/test/java/org/springframework/amqp/support/converter/SerializerMessageConverterTests.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2002-2010 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.support.converter;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
+import org.junit.Test;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Fisher
+ */
+public class SerializerMessageConverterTests {
+
+	@Test
+	public void bytesAsDefaultMessageBodyType() throws Exception {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		Message message = new Message("test".getBytes(), new MessageProperties());
+		Object result = converter.fromMessage(message);
+		assertEquals(byte[].class, result.getClass());
+		assertEquals("test", new String((byte[]) result, "UTF-8"));
+	}
+
+	@Test
+	public void messageToString() {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		Message message = new Message("test".getBytes(), new MessageProperties());
+		message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
+		Object result = converter.fromMessage(message);
+		assertEquals(String.class, result.getClass());
+		assertEquals("test", result);
+	}
+
+	@Test
+	public void messageToBytes() {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		Message message = new Message(new byte[] { 1, 2, 3 }, new MessageProperties());
+		message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_BYTES);
+		Object result = converter.fromMessage(message);
+		assertEquals(byte[].class, result.getClass());
+		byte[] resultBytes = (byte[]) result;
+		assertEquals(3, resultBytes.length);
+		assertEquals(1, resultBytes[0]);
+		assertEquals(2, resultBytes[1]);
+		assertEquals(3, resultBytes[2]);
+	}
+
+	@Test
+	public void messageToSerializedObject() throws Exception {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		MessageProperties properties = new MessageProperties();
+		properties.setContentType(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT);
+		ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+		ObjectOutputStream objectStream = new ObjectOutputStream(byteStream);
+		TestBean testBean = new TestBean("foo");
+		objectStream.writeObject(testBean);
+		objectStream.flush();
+		objectStream.close();
+		byte[] bytes = byteStream.toByteArray();
+		Message message = new Message(bytes, properties);
+		Object result = converter.fromMessage(message);
+		assertEquals(TestBean.class, result.getClass());
+		assertEquals(testBean, result);
+	}
+
+	@Test
+	public void messageToSerializedObjectNoContentType() throws Exception {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		converter.setIgnoreContentType(true);
+		MessageProperties properties = new MessageProperties();
+		ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+		ObjectOutputStream objectStream = new ObjectOutputStream(byteStream);
+		TestBean testBean = new TestBean("foo");
+		objectStream.writeObject(testBean);
+		objectStream.flush();
+		objectStream.close();
+		byte[] bytes = byteStream.toByteArray();
+		Message message = new Message(bytes, properties);
+		Object result = converter.fromMessage(message);
+		assertEquals(TestBean.class, result.getClass());
+		assertEquals(testBean, result);
+	}
+
+	@Test
+	public void stringToMessage() throws Exception {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		Message message = converter.toMessage("test", new MessageProperties());
+		String contentType = message.getMessageProperties().getContentType();
+		String content = new String(message.getBody(),
+				message.getMessageProperties().getContentEncoding());
+		assertEquals("text/plain", contentType);
+		assertEquals("test", content);
+	}
+
+	@Test
+	public void bytesToMessage() throws Exception {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		Message message = converter.toMessage(new byte[] { 1, 2, 3 }, new MessageProperties());
+		String contentType = message.getMessageProperties().getContentType();
+		byte[] body = message.getBody();
+		assertEquals("application/octet-stream", contentType);
+		assertEquals(3, body.length);
+		assertEquals(1, body[0]);
+		assertEquals(2, body[1]);
+		assertEquals(3, body[2]);
+	}
+
+	@Test
+	public void serializedObjectToMessage() throws Exception {
+		SerializerMessageConverter converter = new SerializerMessageConverter();
+		TestBean testBean = new TestBean("foo");
+		Message message = converter.toMessage(testBean, new MessageProperties());
+		String contentType = message.getMessageProperties().getContentType();
+		byte[] body = message.getBody();
+		assertEquals("application/x-java-serialized-object", contentType);
+		ByteArrayInputStream bais = new ByteArrayInputStream(body);
+		Object deserializedObject = new ObjectInputStream(bais).readObject();
+		assertEquals(testBean, deserializedObject);
+	}
+
+
+	@SuppressWarnings("serial")
+	private static class TestBean implements Serializable {
+
+		private final String text;
+
+		TestBean(String text) {
+			Assert.notNull(text, "text must not be null");
+			this.text = text;
+		}
+
+		public boolean equals(Object other) {
+			return (other instanceof TestBean && this.text.equals(((TestBean) other).text));
+		}
+
+		public int hashCode() {
+			return this.text.hashCode();
+		}
+	}
+
+}

spring-amqp-parent/pom.xml

@@ -20,7 +20,7 @@
 		<org.codehaus.jackson.version>1.4.3</org.codehaus.jackson.version>
 		<org.erlang.otp.version>1.5.3</org.erlang.otp.version>
 		<com.rabbitmq.version>2.1.0</com.rabbitmq.version>
-		<org.springframework.version>3.0.3.RELEASE</org.springframework.version>
+		<org.springframework.version>3.0.5.RELEASE</org.springframework.version>
 	</properties>
 	<profiles>
 		<profile>

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -66,10 +66,6 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
 	// The default queue name that will be used for synchronous receives.
 	private volatile String queue;
 
-	private volatile boolean mandatoryPublish;
-
-	private volatile boolean immediatePublish;
-
 	private volatile long replyTimeout = DEFAULT_REPLY_TIMEOUT;
 
 	private volatile MessageConverter messageConverter = new SimpleMessageConverter();
@@ -100,25 +96,6 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
 		this.queue = queue;
 	}
 
-	/**
-	 * If the message doesn't get routed to a queue for any reason, the server will send an async response to let me
-	 * know. Possible use case: check routing.
-	 * 
-	 * @param mandatoryPublish the flag value to set
-	 */
-	public void setMandatoryPublish(boolean mandatoryPublish) {
-		this.mandatoryPublish = mandatoryPublish;
-	}
-
-	/**
-	 * Like a rendezvous.
-	 * 
-	 * @param immediatePublish
-	 */
-	public void setImmediatePublish(boolean immediatePublish) {
-		this.immediatePublish = immediatePublish;
-	}
-
 	/**
 	 * Specify the timeout in milliseconds to be used when waiting for a reply Message when using one of the
 	 * sendAndReceive methods. The default value is defined as {@link #DEFAULT_REPLY_TIMEOUT}. A negative value
@@ -344,8 +321,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
 			routingKey = this.routingKey;
 		}
 		// TODO parameterize out default encoding
-		channel.basicPublish(exchange, routingKey, this.mandatoryPublish, this.immediatePublish,
-				RabbitUtils.extractBasicProperties(message, "UTF-8"), message.getBody());
+		channel.basicPublish(exchange, routingKey, false, false, RabbitUtils.extractBasicProperties(message, "UTF-8"),
+				message.getBody());
 		// Check commit - avoid commit call within a JTA transaction.
 		// TODO: should we be able to do (via wrapper) something like:
 		// channel.getTransacted()?

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

@@ -6,11 +6,11 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.utility.Utility;
 /**
  * Variation on QueueingConsumer in RabbitMQ, uses 'put' instead of 'add' and stored a reference to the consumerTag that
@@ -20,8 +20,6 @@ import com.rabbitmq.utility.Utility;
  */
 public class BlockingQueueConsumer extends DefaultConsumer {
 
-		private String consumerTag;
-	
 	    private final BlockingQueue<Delivery> queue;
 
 	    // When this is non-null the queue is in shutdown mode and nextDelivery should
@@ -43,18 +41,7 @@ public class BlockingQueueConsumer extends DefaultConsumer {
 	        super(ch);
 	        this.queue = q;
 	    }
-	    
-	    
 
-	    public String getConsumerTag() {
-			return consumerTag;
-		}
-	    
-	    public void setConsumerTag(String consumerTag) 
-	    {
-	    	this.consumerTag = consumerTag;
-	    }
-	    
 		@Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
 	        shutdown = sig; 
 	        try {

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

@@ -260,8 +260,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		String[] queue = StringUtils.commaDelimitedListToStringArray(queueNames);
 		for (int i = 0; i < queue.length; i++) {
 			channel.queueDeclarePassive(queue[i]);
-			String consumerTag = channel.basicConsume(queue[i], !isChannelTransacted(), consumer);
-			consumer.setConsumerTag(consumerTag);
+			channel.basicConsume(queue[i], !isChannelTransacted(), consumer);
 		}
 		return consumer;
 	}

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

@@ -3,10 +3,8 @@ package org.springframework.amqp.rabbit.core;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
@@ -23,27 +21,12 @@ import com.rabbitmq.client.Channel;
 
 public class RabbitBindingIntegrationTests {
 
-	private Queue queue;
+	private Queue queue = new Queue("test.queue");
 
-	private RabbitTemplate template = new RabbitTemplate(
-			new CachingConnectionFactory());
+	private RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
-
-	@Before
-	public void declareQueue() {
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue("test.queue");
-		} catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		queue = new Queue("test.queue");
-		// Idempotent, so no problem to do this for every test
-		admin.declareQueue(queue);
-		admin.purgeQueue("test.queue", false);
-	}
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
 
 	@Test
 	public void testSendAndReceiveWithTopicSingleCallback() throws Exception {
@@ -53,25 +36,67 @@ public class RabbitBindingIntegrationTests {
 		admin.declareExchange(exchange);
 		template.setExchange(exchange.getName());
 
-		admin.declareBinding(BindingBuilder.from(queue).to(exchange)
-				.with("*.end"));
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
 
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-						channel);
-				String tag = channel.basicConsume(queue.getName(), true,
-						consumer);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
 				assertNotNull(tag);
 
 				template.convertAndSend("foo", "message");
-				String result = getResult(consumer);
-				assertEquals(null, result);
 
-				template.convertAndSend("foo.end", "message");
-				result = getResult(consumer);
-				assertEquals("message", result);
+				try {
+
+					String result = getResult(consumer);
+					assertEquals(null, result);
+
+					template.convertAndSend("foo.end", "message");
+					result = getResult(consumer);
+					assertEquals("message", result);
+
+				} finally {
+					channel.basicCancel(tag);
+				}
+
+				return null;
+
+			}
+		});
+
+	}
+
+	@Test
+	public void testSendAndReceiveWithNonDefaultExchange() throws Exception {
+
+		final RabbitAdmin admin = new RabbitAdmin(template);
+		final TopicExchange exchange = new TopicExchange("topic");
+		admin.declareExchange(exchange);
+
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
+
+		template.execute(new ChannelCallback<Void>() {
+			public Void doInRabbit(Channel channel) throws Exception {
+
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
+				assertNotNull(tag);
+
+				template.convertAndSend("topic", "foo", "message");
+
+				try {
+
+					String result = getResult(consumer);
+					assertEquals(null, result);
+
+					template.convertAndSend("topic", "foo.end", "message");
+					result = getResult(consumer);
+					assertEquals("message", result);
+
+				} finally {
+					channel.basicCancel(tag);
+				}
 
 				return null;
 
@@ -82,36 +107,29 @@ public class RabbitBindingIntegrationTests {
 
 	@Test
 	// @Ignore("Not sure yet if we need to support a use case like this")
-	public void testSendAndReceiveWithTopicConsumeInBackground()
-			throws Exception {
+	public void testSendAndReceiveWithTopicConsumeInBackground() throws Exception {
 
 		RabbitAdmin admin = new RabbitAdmin(template);
 		TopicExchange exchange = new TopicExchange("topic");
 		admin.declareExchange(exchange);
 		template.setExchange(exchange.getName());
 
-		admin.declareBinding(BindingBuilder.from(queue).to(exchange)
-				.with("*.end"));
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
 
-		final RabbitTemplate template = new RabbitTemplate(
-				new CachingConnectionFactory());
+		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
 		template.setExchange(exchange.getName());
 
-		BlockingQueueConsumer consumer = template
-				.execute(new ChannelCallback<BlockingQueueConsumer>() {
-					public BlockingQueueConsumer doInRabbit(Channel channel)
-							throws Exception {
+		BlockingQueueConsumer consumer = template.execute(new ChannelCallback<BlockingQueueConsumer>() {
+			public BlockingQueueConsumer doInRabbit(Channel channel) throws Exception {
 
-						BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-								channel);
-						String tag = channel.basicConsume(queue.getName(),
-								true, consumer);
-						assertNotNull(tag);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
+				assertNotNull(tag);
 
-						return consumer;
+				return consumer;
 
-					}
-				});
+			}
+		});
 
 		template.convertAndSend("foo", "message");
 		String result = getResult(consumer);
@@ -121,6 +139,8 @@ public class RabbitBindingIntegrationTests {
 		result = getResult(consumer);
 		assertEquals("message", result);
 
+		consumer.getChannel().basicCancel(consumer.getConsumerTag());
+
 	}
 
 	@Test
@@ -131,16 +151,13 @@ public class RabbitBindingIntegrationTests {
 		admin.declareExchange(exchange);
 		template.setExchange(exchange.getName());
 
-		admin.declareBinding(BindingBuilder.from(queue).to(exchange)
-				.with("*.end"));
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
 
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-						channel);
-				String tag = channel.basicConsume(queue.getName(), true,
-						consumer);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
 				assertNotNull(tag);
 
 				try {
@@ -159,10 +176,8 @@ public class RabbitBindingIntegrationTests {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-						channel);
-				String tag = channel.basicConsume(queue.getName(), true,
-						consumer);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
 				assertNotNull(tag);
 
 				try {
@@ -180,15 +195,13 @@ public class RabbitBindingIntegrationTests {
 
 	}
 
-	private String getResult(final BlockingQueueConsumer consumer)
-			throws InterruptedException {
+	private String getResult(final BlockingQueueConsumer consumer) throws InterruptedException {
 		Delivery response = consumer.nextDelivery(200L);
 		if (response == null) {
 			return null;
 		}
-		MessageProperties messageProps = RabbitUtils.createMessageProperties(
-				response.getProperties(), response.getEnvelope(), "UTF-8");
-		return (String) new SimpleMessageConverter().fromMessage(new Message(
-				response.getBody(), messageProps));
+		MessageProperties messageProps = RabbitUtils.createMessageProperties(response.getProperties(),
+				response.getEnvelope(), "UTF-8");
+		return (String) new SimpleMessageConverter().fromMessage(new Message(response.getBody(), messageProps));
 	}
 }

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

@@ -11,14 +11,11 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.UncategorizedAmqpException;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
 import org.springframework.amqp.rabbit.support.RabbitUtils;
@@ -42,20 +39,7 @@ public class RabbitTemplateIntegrationTests {
 	private RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
-
-	@Before
-	public void declareQueue() {
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue(ROUTE);
-		}
-		catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		admin.declareQueue(new Queue(ROUTE));
-		admin.purgeQueue(ROUTE, false);
-	}
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(ROUTE);
 
 	@Test
 	public void testSendAndReceive() throws Exception {

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

@@ -7,8 +7,6 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import org.springframework.amqp.AmqpIOException;
-import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
@@ -30,7 +28,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
 
 	@Rule
 	// After the repeat processor, so it only runs once
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(ROUTE);
 
 	private CachingConnectionFactory connectionFactory;
 
@@ -44,17 +42,6 @@ public class RabbitTemplatePerformanceIntegrationTests {
 		connectionFactory.setChannelCacheSize(repeat.getConcurrency());
 		// connectionFactory.setPort(5673);
 		template.setConnectionFactory(connectionFactory);
-		// TODO: investigate the effects of these flags...
-		// template.setMandatoryPublish(true);
-		// template.setImmediatePublish(true);
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue(ROUTE);
-		} catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		admin.declareQueue(new Queue(ROUTE));
-		admin.purgeQueue(ROUTE, false);
 	}
 
 	@After

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java

@@ -18,10 +18,8 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
@@ -31,7 +29,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerContainerLifecycleIntegrationTests.class);
 
-	private Queue queue;
+	private Queue queue = new Queue("test.queue");
 
 	private RabbitTemplate template = new RabbitTemplate();
 
@@ -40,7 +38,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 	private final boolean transactional;
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
 
 	private final int messageCount;
 
@@ -62,17 +60,6 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 		connectionFactory.setChannelCacheSize(concurrentConsumers);
 		connectionFactory.setPort(5673);
 		template.setConnectionFactory(connectionFactory);
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue("test.queue");
-		}
-		catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		queue = new Queue("test.queue");
-		// Idempotent, so no problem to do this for every test
-		admin.declareQueue(queue);
-		admin.purgeQueue("test.queue", false);
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java

@@ -20,10 +20,8 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
@@ -45,7 +43,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		}
 	}
 
-	private Queue queue;
+	private Queue queue = new Queue("test.queue");
 
 	private RabbitTemplate template = new RabbitTemplate();
 
@@ -58,16 +56,20 @@ public class SimpleMessageListenerContainerIntegrationTests {
 			SimpleMessageListenerContainer.class);
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
 
 	private final int messageCount;
 
 	private SimpleMessageListenerContainer container;
 
-	public SimpleMessageListenerContainerIntegrationTests(int messageCount, int concurrency, TransactionType transacted) {
+	private final int txSize;
+
+	public SimpleMessageListenerContainerIntegrationTests(int messageCount, int concurrency,
+			TransactionType transacted, int txSize) {
 		this.messageCount = messageCount;
 		this.concurrentConsumers = concurrency;
 		this.transactional = transacted;
+		this.txSize = txSize;
 	}
 
 	@Parameters
@@ -76,11 +78,16 @@ public class SimpleMessageListenerContainerIntegrationTests {
 				params(2, 4, 1, TransactionType.NATIVE), params(3, 4, 1, TransactionType.EXTERNAL),
 				params(4, 2, 2, TransactionType.NATIVE), params(5, 2, 2, TransactionType.NONE),
 				params(6, 20, 4, TransactionType.NATIVE), params(7, 20, 4, TransactionType.NONE),
-				params(8, 1000, 4, TransactionType.NATIVE), params(9, 1000, 4, TransactionType.NONE));
+				params(8, 1000, 4, TransactionType.NATIVE), params(9, 1000, 4, TransactionType.NONE),
+				params(10, 1000, 4, TransactionType.NATIVE, 10));
+	}
+
+	private static Object[] params(int i, int messageCount, int concurrency, TransactionType transacted, int txSize) {
+		return new Object[] { messageCount, concurrency, transacted, txSize };
 	}
 
 	private static Object[] params(int i, int messageCount, int concurrency, TransactionType transacted) {
-		return new Object[] { messageCount, concurrency, transacted };
+		return params(i, messageCount, concurrency, transacted, 1);
 	}
 
 	@Before
@@ -89,16 +96,6 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		connectionFactory.setChannelCacheSize(concurrentConsumers);
 		// connectionFactory.setPort(5673);
 		template.setConnectionFactory(connectionFactory);
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue("test.queue");
-		} catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		queue = new Queue("test.queue");
-		// Idempotent, so no problem to do this for every test
-		admin.declareQueue(queue);
-		admin.purgeQueue("test.queue", false);
 	}
 
 	@After
@@ -156,7 +153,8 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory());
 		container.setMessageListener(new MessageListenerAdapter(listener));
 		container.setQueueName(queue.getName());
-		container.setPrefetchCount(1);
+		container.setTxSize(txSize);
+		container.setPrefetchCount(txSize);
 		container.setConcurrentConsumers(concurrentConsumers);
 		container.setChannelTransacted(transactional.isTransactional());
 		if (transactional == TransactionType.EXTERNAL) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java

@@ -24,12 +24,12 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
  * 
  * @Test
  * public void testSendAndReceive() throws Exception {
- * 	// ... test using RabbitTemplate etc. 
+ * 	// ... test using RabbitTemplate etc.
  * }
  * </pre>
  * <p>
- * It is recommended to declare the rule as static so that it only has to check once for all tests in the enclosing test
- * case.
+ * The rule can be declared as static so that it only has to check once for all tests in the enclosing test case, but
+ * there isn't a lot of overhead in making it non-static.
  * </p>
  * 
  * @see Assume
@@ -40,6 +40,8 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
  */
 public class BrokerRunning extends TestWatchman {
 
+	private static final String DEFAULT_QUEUE_NAME = BrokerRunning.class.getName();
+
 	private static Log logger = LogFactory.getLog(BrokerRunning.class);
 
 	private boolean brokerOnline = true;
@@ -48,6 +50,28 @@ public class BrokerRunning extends TestWatchman {
 
 	private final boolean assumeOnline;
 
+	private final boolean purge;
+
+	private Queue queue;
+
+	/**
+	 * Ensure the broker is running and has an empty queue with the specified name in the default exchange.
+	 * 
+	 * @return a new rule that assumes an existing running broker
+	 */
+	public static BrokerRunning isRunningWithEmptyQueue(String queue) {
+		return new BrokerRunning(true, new Queue(queue), true);
+	}
+
+	/**
+	 * Ensure the broker is running and has an empty queue in the default exchange.
+	 * 
+	 * @return a new rule that assumes an existing running broker
+	 */
+	public static BrokerRunning isRunningWithEmptyQueue(Queue queue) {
+		return new BrokerRunning(true, queue, true);
+	}
+
 	/**
 	 * @return a new rule that assumes an existing running broker
 	 */
@@ -62,13 +86,24 @@ public class BrokerRunning extends TestWatchman {
 		return new BrokerRunning(false);
 	}
 
-	private BrokerRunning(boolean assumeOnline) {
+	private BrokerRunning(boolean assumeOnline, Queue queue, boolean purge) {
 		this.assumeOnline = assumeOnline;
+		this.queue = queue;
+		this.purge = purge;
+	}
+
+	private BrokerRunning(boolean assumeOnline, Queue queue) {
+		this(assumeOnline, queue, false);
+	}
+
+	private BrokerRunning(boolean assumeOnline) {
+		this(assumeOnline, new Queue(DEFAULT_QUEUE_NAME));
 	}
 
 	@Override
 	public Statement apply(Statement base, FrameworkMethod method, Object target) {
 
+		// Check at the beginning, so this can be used as a static field
 		if (assumeOnline) {
 			Assume.assumeTrue(brokerOnline);
 		} else {
@@ -76,14 +111,30 @@ public class BrokerRunning extends TestWatchman {
 		}
 
 		try {
+
 			CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 			RabbitAdmin admin = new RabbitAdmin(connectionFactory);
-			admin.declareQueue(new Queue("test.broker.running"));
-			admin.deleteQueue("test.broker.running");
+
+			String queueName = queue.getName();
+
+			if (purge) {
+				logger.debug("Deleting queue: " + queueName);
+				// Delete completely - gets rid of consumers and bindings as well
+				admin.deleteQueue(queueName);
+			}
+
+			admin.declareQueue(queue);
+
+			if (isDefaultQueue(queueName)) {
+				// Just for test probe.
+				admin.deleteQueue(queueName);
+				queue = null;
+			}
 			brokerOffline = false;
 			if (!assumeOnline) {
 				Assume.assumeTrue(brokerOffline);
 			}
+
 		} catch (Exception e) {
 			logger.warn("Not executing tests because basic connectivity test failed", e);
 			brokerOnline = false;
@@ -92,9 +143,12 @@ public class BrokerRunning extends TestWatchman {
 			}
 		}
 
-
 		return super.apply(base, method, target);
 
 	}
 
+	private boolean isDefaultQueue(String queue) {
+		return DEFAULT_QUEUE_NAME.equals(queue);
+	}
+
 }