- * (exchangeType)://(exchange)/(routingKey) + * (exchange)/(routingKey) ** - * Here we also allow the exchange type to default to direct, and the exchange name to default to empty (so just a - * routing key will work if you know the queue name). - * - * @see ExchangeTypes + * Here we also the exchange name to default to empty + * (so just a routing key will work if you know the queue name). * * @author Mark Pollack * @author Mark Fisher * @author Dave Syer + * @author Artem Bilan */ public class Address { - private static final Pattern pattern = Pattern.compile("^([^:]+)://([^/]*)/?(.*)$"); - - private final String exchangeType; + private static final Pattern pattern = Pattern.compile("^(?:.*://)?([^/]*)/?(.*)$"); private final String exchangeName; @@ -50,28 +48,28 @@ public class Address { * Create an Address instance from a structured String in the form * *
- * (exchangeType)://(exchange)/(routingKey) + * (exchange)/(routingKey) ** - * where examples of valid
exchangeType values can be found in the {@link ExchangeTypes} static
- * constants.
- *
* @param address a structured string.
*/
public Address(String address) {
- if (address == null) {
- this.exchangeType = ExchangeTypes.DIRECT;
+ if (!StringUtils.hasText(address)) {
this.exchangeName = "";
this.routingKey = "";
- } else {
+ }
+ else if (address.lastIndexOf('/') <= 0) {
+ this.routingKey = address.replaceFirst("/", "");
+ this.exchangeName = "";
+ }
+ else {
Matcher matcher = pattern.matcher(address);
boolean matchFound = matcher.find();
if (matchFound) {
- this.exchangeType = matcher.group(1);
- this.exchangeName = matcher.group(2);
- this.routingKey = matcher.group(3);
- } else {
- this.exchangeType = ExchangeTypes.DIRECT;
+ this.exchangeName = matcher.group(1);
+ this.routingKey = matcher.group(2);
+ }
+ else {
this.exchangeName = "";
this.routingKey = address;
}
@@ -79,21 +77,32 @@ public class Address {
}
/***
- * Create an Address given the exchange type, exchange name and routing key. This will set the exchange type, name
- * and the routing key explicitly.
- *
+ * Create an Address given the exchange type, exchange name and routing key.
+ * This will set the exchange type, name and the routing key explicitly.
* @param exchangeType The exchange type.
* @param exchangeName The exchange name.
* @param routingKey The routing key.
*/
+ @Deprecated
public Address(String exchangeType, String exchangeName, String routingKey) {
- this.exchangeType = exchangeType;
this.exchangeName = exchangeName;
this.routingKey = routingKey;
}
+ /***
+ * Create an Address given the exchange name and routing key.
+ * This will set the exchange type, name and the routing key explicitly.
+ * @param exchangeName The exchange name.
+ * @param routingKey The routing key.
+ */
+ public Address(String exchangeName, String routingKey) {
+ this.exchangeName = exchangeName;
+ this.routingKey = routingKey;
+ }
+
+ @Deprecated
public String getExchangeType() {
- return this.exchangeType;
+ return null;
}
public String getExchangeName() {
@@ -106,7 +115,7 @@ public class Address {
@Override
public String toString() {
- StringBuilder sb = new StringBuilder(this.exchangeType + "://" + this.exchangeName + "/");
+ StringBuilder sb = new StringBuilder(this.exchangeName + "/");
if (StringUtils.hasText(this.routingKey)) {
sb.append(this.routingKey);
}
diff --git a/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java b/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java
index 316687a5..07578b51 100644
--- a/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java
+++ b/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java
@@ -24,13 +24,14 @@ import org.junit.Test;
/**
* @author Mark Pollack
* @author Mark Fisher
+ * @author Artem Bilan
*/
public class AddressTests {
@Test
public void toStringCheck() {
- Address address = new Address(ExchangeTypes.DIRECT, "my-exchange", "routing-key");
- String replyToUri = "direct://my-exchange/routing-key";
+ Address address = new Address("my-exchange", "routing-key");
+ String replyToUri = "my-exchange/routing-key";
Assert.assertEquals(replyToUri, address.toString());
}
@@ -38,34 +39,48 @@ public class AddressTests {
public void parse() {
String replyToUri = "direct://my-exchange/routing-key";
Address address = new Address(replyToUri);
- assertEquals(address.getExchangeType(), ExchangeTypes.DIRECT);
- assertEquals(address.getExchangeName(), "my-exchange");
- assertEquals(address.getRoutingKey(), "routing-key");
+ assertEquals("my-exchange", address.getExchangeName());
+ assertEquals("routing-key", address.getRoutingKey());
}
@Test
public void parseUnstructuredWithRoutingKeyOnly() {
Address address = new Address("my-routing-key");
assertEquals("my-routing-key", address.getRoutingKey());
- assertEquals("direct:///my-routing-key", address.toString());
+ assertEquals("/my-routing-key", address.toString());
+
+ address = new Address("/foo");
+ assertEquals("foo", address.getRoutingKey());
+ assertEquals("/foo", address.toString());
+
+ address = new Address("bar/baz");
+ assertEquals("bar", address.getExchangeName());
+ assertEquals("baz", address.getRoutingKey());
+ assertEquals("bar/baz", address.toString());
}
@Test
public void parseWithoutRoutingKey() {
Address address = new Address("fanout://my-exchange");
- assertEquals(ExchangeTypes.FANOUT, address.getExchangeType());
assertEquals("my-exchange", address.getExchangeName());
assertEquals("", address.getRoutingKey());
- assertEquals("fanout://my-exchange/", address.toString());
+ assertEquals("my-exchange/", address.toString());
}
@Test
public void parseWithDefaultExchangeAndRoutingKey() {
Address address = new Address("direct:///routing-key");
- assertEquals(ExchangeTypes.DIRECT, address.getExchangeType());
assertEquals("", address.getExchangeName());
assertEquals("routing-key", address.getRoutingKey());
- assertEquals("direct:///routing-key", address.toString());
+ assertEquals("/routing-key", address.toString());
+ }
+
+ @Test
+ public void testEmpty() {
+ Address address = new Address("/");
+ assertEquals("", address.getExchangeName());
+ assertEquals("", address.getRoutingKey());
+ assertEquals("/", address.toString());
}
}
diff --git a/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java b/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java
index bf078ace..3604434b 100644
--- a/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java
+++ b/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java
@@ -29,7 +29,7 @@ public class MessagePropertiesTests {
@Test
public void testReplyTo() throws Exception {
MessageProperties properties = new MessageProperties();
- properties.setReplyTo("fanout://foo/bar");
+ properties.setReplyTo("foo/bar");
assertEquals("bar", properties.getReplyToAddress().getRoutingKey());
}
diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java
index e3228089..8c200413 100644
--- a/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java
+++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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
@@ -61,7 +61,7 @@ public class RemotingTest {
public Object convertSendAndReceive(Object payload) throws AmqpException {
MessageConverter messageConverter = serviceExporter.getMessageConverter();
- Address replyTo = new Address("fakeExchange", "fakeExchangeName", "fakeRoutingKey");
+ Address replyTo = new Address("fakeExchangeName", "fakeRoutingKey");
MessageProperties messageProperties = new MessageProperties();
messageProperties.setReplyToAddress(replyTo);
Message message = messageConverter.toMessage(payload, messageProperties);
@@ -89,6 +89,7 @@ public class RemotingTest {
}
@Test(expected = GeneralException.class)
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
public void testExceptionReturningMethod() {
riggedProxy.notReallyExceptionReturningMethod();
}
@@ -96,8 +97,6 @@ public class RemotingTest {
@Test
public void testActuallyExceptionReturningMethod() {
SpecialException returnedException = riggedProxy.actuallyExceptionReturningMethod();
-
Assert.assertNotNull(returnedException);
- Assert.assertTrue(returnedException instanceof SpecialException);
}
}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
index 984ebeac..29f3581d 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
@@ -118,16 +118,6 @@ public @interface RabbitListener {
*/
String priority() default "";
- /**
- * The routing key to send along with a response message.
- * This will be applied in case of a request message that does not carry
- * a "replyTo" property. Note: This only applies to a listener method with
- * a return value, for which each result object will be converted into a
- * response message.
- * @return the response routing key.
- */
- String responseRoutingKey() default "";
-
/**
* Reference to a {@link org.springframework.amqp.rabbit.core.RabbitAdmin
* RabbitAdmin}. Required if the listener is using auto-delete
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
index 5a998e7c..76bd7834 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
@@ -237,9 +237,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
rabbitListener + " (must be an integer)", ex);
}
}
- if (StringUtils.hasText(rabbitListener.responseRoutingKey())) {
- endpoint.setResponseRoutingKey(resolve(rabbitListener.responseRoutingKey()));
- }
+
String rabbitAdmin = resolve(rabbitListener.admin());
if (StringUtils.hasText(rabbitAdmin)) {
Assert.state(this.beanFactory != null, "BeanFactory must be set to resolve RabbitAdmin by bean name");
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java
index 0e45995c..421c6ec5 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java
@@ -27,7 +27,6 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
-import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener;
import org.springframework.util.Assert;
/**
@@ -50,8 +49,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
private Integer priority;
- private String responseRoutingKey;
-
private RabbitAdmin admin;
@@ -135,21 +132,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
return priority;
}
- /**
- * Set the routing key to send along with a response message.
- * @param responseRoutingKey the response routing key value.
- */
- public void setResponseRoutingKey(String responseRoutingKey) {
- this.responseRoutingKey = responseRoutingKey;
- }
-
- /**
- * @return the routing key to send along with a response message.
- */
- public String getResponseRoutingKey() {
- return responseRoutingKey;
- }
-
/**
* Set the {@link RabbitAdmin} instance to use.
* @param admin the {@link RabbitAdmin} instance.
@@ -207,9 +189,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
private void setupMessageListener(MessageListenerContainer container) {
MessageListener messageListener = createMessageListener(container);
- if (getResponseRoutingKey() != null && messageListener instanceof AbstractAdaptableMessageListener) {
- ((AbstractAdaptableMessageListener) messageListener).setResponseRoutingKey(getResponseRoutingKey());
- }
Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener");
container.setupMessageListener(messageListener);
}
@@ -225,7 +204,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
append("' | queueNames='").append(this.queueNames).
append("' | exclusive='").append(this.exclusive).
append("' | priority='").append(this.priority).
- append("' | responseRoutingKey='").append(this.responseRoutingKey).
append("' | admin='").append(this.admin).append("'");
}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java
index d3f2fe4d..196d073e 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2014 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.
@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.config;
import java.lang.reflect.Method;
import java.util.Arrays;
+import org.springframework.amqp.core.Address;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.amqp.support.converter.MessageConverter;
@@ -27,13 +28,13 @@ import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
/**
* A {@link RabbitListenerEndpoint} providing the method to invoke to process
* an incoming message for this endpoint.
*
* @author Stephane Nicoll
+ * @author Artem Bilan
* @since 1.4
*/
public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint {
@@ -88,9 +89,10 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
InvocableHandlerMethod invocableHandlerMethod =
this.messageHandlerMethodFactory.createInvocableHandlerMethod(getBean(), getMethod());
messageListener.setHandlerMethod(invocableHandlerMethod);
- String responseExchange = getDefaultResponseExchange();
- if (StringUtils.hasText(responseExchange)) {
- messageListener.setResponseExchange(responseExchange);
+ Address replyToAddress = getDefaultReplyToAddress();
+ if (replyToAddress != null) {
+ messageListener.setResponseExchange(replyToAddress.getExchangeName());
+ messageListener.setResponseRoutingKey(replyToAddress.getRoutingKey());
}
MessageConverter messageConverter = container.getMessageConverter();
if (messageConverter != null) {
@@ -107,15 +109,15 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
return new MessagingMessageListenerAdapter();
}
- private String getDefaultResponseExchange() {
+ private Address getDefaultReplyToAddress() {
SendTo ann = AnnotationUtils.getAnnotation(getMethod(), SendTo.class);
if (ann != null) {
- Object[] destinations = ann.value();
- if (destinations.length != 1) {
+ String[] destinations = ann.value();
+ if (destinations.length > 1) {
throw new IllegalStateException("Invalid @" + SendTo.class.getSimpleName() + " annotation on '"
+ getMethod() + "' one destination must be set (got " + Arrays.toString(destinations) + ")");
}
- return (String) destinations[0];
+ return destinations.length == 1 ? new Address(destinations[0]) : new Address(null);
}
return null;
}
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 292cdd6a..1ba602b1 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
@@ -544,7 +544,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
@Override
public Address getReplyToAddress(Message request, S reply) {
- return new Address(null, replyExchange, replyRoutingKey);
+ return new Address(replyExchange, replyRoutingKey);
}
});
@@ -976,7 +976,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
"Cannot determine ReplyTo message property value: "
+ "Request message does not contain reply-to property, and no default Exchange was set.");
}
- replyTo = new Address(null, this.exchange, this.routingKey);
+ replyTo = new Address(this.exchange, this.routingKey);
}
return replyTo;
}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
index 99d39f48..9fe11a2e 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
@@ -280,7 +280,7 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene
"Request message does not contain reply-to property, " +
"and no default response Exchange was set.");
}
- replyTo = new Address(null, this.responseExchange, this.responseRoutingKey);
+ replyTo = new Address(this.responseExchange, this.responseRoutingKey);
}
return replyTo;
}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java
index f9c0fc5b..0c234409 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java
@@ -16,9 +16,15 @@
package org.springframework.amqp.rabbit.annotation;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
import java.util.Collection;
-import com.rabbitmq.client.Channel;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -40,8 +46,7 @@ import org.springframework.validation.Errors;
import org.springframework.validation.Validator;
import org.springframework.validation.annotation.Validated;
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
+import com.rabbitmq.client.Channel;
/**
*
@@ -117,7 +122,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
assertTrue("No queue instances should be set", endpoint.getQueues().isEmpty());
assertEquals(true, endpoint.isExclusive());
assertEquals(new Integer(34), endpoint.getPriority());
- assertEquals("routing-123", endpoint.getResponseRoutingKey());
assertSame(context.getBean("rabbitAdmin"), endpoint.getAdmin());
// Resolve the container and invoke a message on it
@@ -142,7 +146,7 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
static class FullBean {
@RabbitListener(id = "listener1", containerFactory = "simpleFactory", queues = {"queue1", "queue2"},
- exclusive = true, priority = "34", responseRoutingKey = "routing-123", admin = "rabbitAdmin")
+ exclusive = true, priority = "34", admin = "rabbitAdmin")
public void fullHandle(String msg) {
}
@@ -153,8 +157,7 @@ public abstract class AbstractRabbitAnnotationDrivenTests {
@RabbitListener(id = "${rabbit.listener.id}", containerFactory = "${rabbit.listener.containerFactory}",
queues = {"${rabbit.listener.queue}", "queue2"}, exclusive = true,
- priority = "${rabbit.listener.priority}", responseRoutingKey = "${rabbit.listener.responseRoutingKey}",
- admin = "${rabbit.listener.admin}")
+ priority = "${rabbit.listener.priority}", admin = "${rabbit.listener.admin}")
public void fullHandle(String msg) {
}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
index f74c6744..c37bd0c1 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2014 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.
@@ -16,6 +16,10 @@
package org.springframework.amqp.rabbit.annotation;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -33,16 +37,17 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
-import static org.junit.Assert.*;
-
/**
*
* @author Stephane Nicoll
+ * @author Artem Bilan
+ * @since 1.4
*/
@ContextConfiguration(classes = EnableRabbitIntegrationTests.EnableRabbitConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@@ -51,7 +56,7 @@ public class EnableRabbitIntegrationTests {
@ClassRule
public static final BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues(
- "test.simple", "test.header", "test.message", "test.reply");
+ "test.simple", "test.header", "test.message", "test.reply", "test.sendTo", "test.sendTo.reply");
@Autowired
private RabbitTemplate rabbitTemplate;
@@ -90,6 +95,19 @@ public class EnableRabbitIntegrationTests {
assertEquals("Wrong bar header", "barValue", reply.getMessageProperties().getHeaders().get("bar"));
}
+ @Test
+ public void simpleEndpointWithSendTo() throws InterruptedException {
+ rabbitTemplate.convertAndSend("test.sendTo", "bar");
+ int n = 0;
+ Object result = null;
+ while ((result = rabbitTemplate.receiveAndConvert("test.sendTo.reply")) == null && n++ < 10) {
+ Thread.sleep(100);
+ }
+ assertTrue(n < 10);
+ assertNotNull(result);
+ assertEquals("BAR", result);
+ }
+
public static class MyService {
@RabbitListener(queues = "test.simple")
@@ -112,6 +130,12 @@ public class EnableRabbitIntegrationTests {
return MessageBuilder.withPayload(payload)
.setHeader("foo", foo).setHeader("bar", "barValue").build();
}
+
+ @RabbitListener(queues = "test.sendTo")
+ @SendTo("test.sendTo.reply")
+ public String capitalizeAndSendTo(String foo) {
+ return foo.toUpperCase();
+ }
}
@Configuration
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java
index 75592cab..78a677ef 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java
@@ -16,14 +16,22 @@
package org.springframework.amqp.rabbit.config;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.AdditionalMatchers.aryEq;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.springframework.amqp.rabbit.test.MessageTestUtils.createTextMessage;
+
import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Rule;
@@ -32,7 +40,6 @@ import org.junit.rules.ExpectedException;
import org.junit.rules.TestName;
import org.mockito.ArgumentCaptor;
-import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.ListenerExecutionFailedException;
@@ -59,11 +66,8 @@ import org.springframework.validation.Errors;
import org.springframework.validation.Validator;
import org.springframework.validation.annotation.Validated;
-import static org.junit.Assert.*;
-import static org.mockito.AdditionalMatchers.*;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.*;
-import static org.springframework.amqp.rabbit.test.MessageTestUtils.*;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
/**
@@ -275,7 +279,7 @@ public class MethodRabbitListenerEndpointTests {
MessagingMessageListenerAdapter listener = createDefaultInstance(String.class);
listener.setMandatoryPublish(true);
String body = "echo text";
- Address replyTo = new Address(null, "replyToQueue", "myRouting");
+ Address replyTo = new Address("replyToQueue", "myRouting");
MessageProperties properties = new MessageProperties();
properties.setReplyToAddress(replyTo);
@@ -319,11 +323,13 @@ public class MethodRabbitListenerEndpointTests {
public void emptySendTo() throws Exception {
MessagingMessageListenerAdapter listener = createDefaultInstance(String.class);
- Channel channel = mock(Channel.class);
+ processAndReply(listener, createTextMessage("content"), "", "", false, null);
+ assertDefaultListenerMethodInvocation();
+ }
- thrown.expect(ReplyFailureException.class);
- thrown.expectCause(Matchers.isA(AmqpException.class));
- listener.onMessage(createTextMessage("content"), channel);
+ @Test
+ public void noSendToValue() throws Exception {
+ emptySendTo();
}
@Test
@@ -392,8 +398,7 @@ public class MethodRabbitListenerEndpointTests {
endpoint.setBean(sample);
endpoint.setMethod(method);
endpoint.setMessageHandlerMethodFactory(factory);
- MessagingMessageListenerAdapter messageListener = endpoint.createMessageListener(container);
- return messageListener;
+ return endpoint.createMessageListener(container);
}
private MessagingMessageListenerAdapter createInstance(
@@ -533,7 +538,7 @@ public class MethodRabbitListenerEndpointTests {
return content;
}
- @SendTo("replyDestination")
+ @SendTo("replyDestination/")
public String processAndReplyWithSendTo(String content) {
invocations.put("processAndReplyWithSendTo", true);
return content;
@@ -545,6 +550,12 @@ public class MethodRabbitListenerEndpointTests {
return content;
}
+ @SendTo
+ public String noSendToValue(String content) {
+ invocations.put("noSendToValue", true);
+ return content;
+ }
+
@SendTo({"firstDestination", "secondDestination"})
public String invalidSendTo(String content) {
invocations.put("invalidSendTo", true);
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 bf67c424..d311610e 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
@@ -856,7 +856,7 @@ public class RabbitTemplateIntegrationTests {
public Message handle(Message message) {
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType(message.getMessageProperties().getContentType());
- messageProperties.setHeader("testReplyTo", new Address("", "", ROUTE));
+ messageProperties.setHeader("testReplyTo", new Address("", ROUTE));
return new Message(message.getBody(), messageProperties);
}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
index 2a78daca..4d3b1f0c 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
@@ -743,35 +743,4 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
assertNull(templateWithConfirmsEnabled.getUnconfirmed(0));
}
- @Test
- public void testReturnNotReceivedAfterPublisherCallbackChannelClose() throws Exception {
- final CountDownLatch latch = new CountDownLatch(20);
- templateWithReturnsEnabled.setMandatory(true);
- templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() {
-
- @Override
- public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
- latch.countDown();
- }
-
- });
-
- ExecutorService executorService = Executors.newCachedThreadPool();
- for (int i = 0; i < 20; i++) {
- executorService.execute(new Runnable() {
-
- @Override
- public void run() {
- templateWithReturnsEnabled.convertAndSend("BAD_ROUTE", (Object) "bad", new CorrelationData("cba"));
- }
-
- });
- }
-
- executorService.shutdown();
- assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
- Thread.sleep(100);
- assertFalse(latch.getCount() == 0);
- }
-
}
diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml
index ec2a8dc9..58fa75ec 100644
--- a/src/reference/docbook/amqp.xml
+++ b/src/reference/docbook/amqp.xml
@@ -1250,6 +1250,28 @@ public Messageexchange/routingKey, where one of those
+ parts can be omitted. The valid values are:
+ foo/bar - the replyTo exchange and routingKey.
+ foo/ - the replyTo exchange and default (empty) routingKey.
+ bar or /bar - the replyTo routingKey and default (empty) exchange.
+ / or empty - the replyTo default exchange and default routingKey.
+ value attribute. This case is equal to an empty
+ sendTo pattern.