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);
+ }
+
+}
diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java b/spring-amqp-core/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java
index 1f92a424..cb23b258 100644
--- a/spring-amqp-core/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java
+++ b/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.
+ *
+ * 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 null)
+ * @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);
+ }
+
}
diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/utils/SerializationUtils.java b/spring-amqp-core/src/main/java/org/springframework/amqp/utils/SerializationUtils.java
index 10bb210f..35434d5d 100644
--- a/spring-amqp-core/src/main/java/org/springframework/amqp/utils/SerializationUtils.java
+++ b/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);
diff --git a/spring-amqp-core/src/test/java/org/springframework/amqp/support/converter/SerializerMessageConverterTests.java b/spring-amqp-core/src/test/java/org/springframework/amqp/support/converter/SerializerMessageConverterTests.java
new file mode 100644
index 00000000..e8e0f548
--- /dev/null
+++ b/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();
+ }
+ }
+
+}
diff --git a/spring-amqp-parent/pom.xml b/spring-amqp-parent/pom.xml
index 1415fcc0..373ae3f4 100644
--- a/spring-amqp-parent/pom.xml
+++ b/spring-amqp-parent/pom.xml
@@ -20,7 +20,7 @@
1.4.3
1.5.3
2.1.0
- 3.0.3.RELEASE
+ 3.0.5.RELEASE
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
index 0e9b8b0f..091c5a91 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
+++ b/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()?
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
index 832f3a6f..8931c575 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
+++ b/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 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 {
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
index 73a05e0c..a503440a 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
+++ b/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;
}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
index db0ba65e..6ff226a0 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
+++ b/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() {
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() {
+ 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() {
- public BlockingQueueConsumer doInRabbit(Channel channel)
- throws Exception {
+ BlockingQueueConsumer consumer = template.execute(new ChannelCallback() {
+ 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() {
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() {
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));
}
}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
index 891cc8d9..fb02d90c 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
+++ b/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 {
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java
index 58e410d9..5b20d196 100755
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java
+++ b/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
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java
index 438882c0..8726a372 100755
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java
+++ b/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
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java
index 52395464..9fc3b24a 100755
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java
+++ b/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) {
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java
index 53c8e951..a0751b24 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java
+++ b/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.
* }
*
*
- * 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.
*
*
* @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);
+ }
+
}