AMQP-429: Change @SendTo Logic to Address

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

Since `@SendTo` can apply only `String` values, but from other side `Address` can parse that `String`,
there is no stops to use `Address` object as a `replyToAddress`.
Now to send a result of `@RabbitListener` method to some specific `replyTo` you can specify it like this:
```
@RabbitListener(queues = "test.sendTo")
@SendTo("direct:///test.sendTo.reply")
public String capitalizeAndSendTo(String foo) {
	return foo.toUpperCase();
}
```

* Add integration test
* In addition remove redundant `@RabbitListener.responseRoutingKey()`

AMQP-429: Revert a legacy code

Note: `@SendTo("")` is a valid use case, when we just want to send message to the Default Exchange with default `routingKey`.
Hence fix `MethodRabbitListenerEndpointTests#emptySendTo()` to expect valid result instead of `Exception`
Fix `Address` to treat empty String as `null`

AMQP-429: Pushed the change for empty `@SendTo`

AMQP-429: Fix `Address` overhead with `ExchangeType`

* Provide new `Address` algorithm to parse String argument
* Fix tests
* Add note to Docs regarding `@SendTo`

AMQP-429: Provide backward compatibility for `Address`

Remove redundant test for `returns`

Polishing

spring-amqp/src/main/java/org/springframework/amqp/core/Address.java

@@ -19,28 +19,26 @@ import java.util.regex.Pattern;
 import org.springframework.util.StringUtils;
 
 /**
- * Represents an address for publication of an AMQP message. The AMQP 0-8 and 0-9 specifications have an unstructured
- * string that is used as a "reply to" address. There are however conventions in use and this class makes it easier to
+ * Represents an address for publication of an AMQP message. The AMQP 0-8 and 0-9
+ * specifications have an unstructured string that is used as a "reply to" address.
+ * There are however conventions in use and this class makes it easier to
  * follow these conventions, which can be easily summarised as:
  *
  * <pre class="code">
- * (exchangeType)://(exchange)/(routingKey)
+ * (exchange)/(routingKey)
  * </pre>
  *
- * 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
 	 *
 	 * <pre class="code">
-	 * (exchangeType)://(exchange)/(routingKey)
+	 * (exchange)/(routingKey)
 	 * </pre>
 	 *
-	 * where examples of valid <code>exchangeType</code> 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);
 		}

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());
 	}
 
 }

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());
 	}
 

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);
 	}
 }

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.
-	 * <p>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

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");

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("'");
 	}
 

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;
 	}

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;
 	}

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;
 	}

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) {
 
 		}

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

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);

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);
 			}
 

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);
-	}
-
 }

src/reference/docbook/amqp.xml

@@ -1250,6 +1250,28 @@ public Message<OrderStatus> processOrder(Order order) {
         .setHeader("code", 1234)
         .build();
 }]]></programlisting>
+	<para>
+		The <literal>@SendTo</literal> value is assumed as a reply <literal>exchange</literal> and
+		<literal>routingKey</literal> pair following the pattern <code>exchange/routingKey</code>, where one of those
+		parts can be omitted. The valid values are:
+		<itemizedlist>
+			<listitem>
+				<code>foo/bar</code> - the replyTo exchange and routingKey.
+			</listitem>
+			<listitem>
+				<code>foo/</code> - the replyTo exchange and default (empty) routingKey.
+			</listitem>
+			<listitem>
+				<code>bar</code> or <code>/bar</code> - the replyTo routingKey and default (empty) exchange.
+			</listitem>
+			<listitem>
+				<code>/</code> or empty - the replyTo default exchange and default routingKey.
+			</listitem>
+		</itemizedlist>
+		Also <literal>@SendTo</literal> can be used without a <code>value</code> attribute. This case is equal to an empty
+		sendTo pattern. <literal>@SendTo</literal> is only used if the inbound message does not have a
+		<literal>replyToAddress</literal> property.
+	</para>
         </section>
     </section>
   </section>