AMQP-466: Move Direct reply-to Decoding to Address

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

1.4.1 introduced direct reply to and reply to address decoding required
using `AddressUtils`. The code should have been put in the `Address` ctor.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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
@@ -44,6 +44,8 @@ public class Address {
 
 	private final String routingKey;
 
+	public static final String AMQ_RABBITMQ_REPLY_TO = "amq.rabbitmq.reply-to";
+
 	/**
 	 * Create an Address instance from a structured String in the form
 	 *
@@ -58,6 +60,10 @@ public Address(String address) {
 			this.exchangeName = "";
 			this.routingKey = "";
 		}
+		else if (address.startsWith(AMQ_RABBITMQ_REPLY_TO)) {
+			this.routingKey = address;
+			this.exchangeName = "";
+		}
 		else if (address.lastIndexOf('/') <= 0) {
 			this.routingKey = address.replaceFirst("/", "");
 			this.exchangeName = "";

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2015 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.
@@ -22,27 +22,14 @@
  */
 public class AddressUtils {
 
-	public static final String AMQ_RABBITMQ_REPLY_TO = "amq.rabbitmq.reply-to";
-
 	/**
 	 * Decodes the reply-to {@link Address} into exchange/key.
 	 *
 	 * @param request the inbound message.
 	 * @return the Address.
 	 */
 	public static Address decodeReplyToAddress(Message request) {
-		Address replyTo;
-		String replyToString = request.getMessageProperties().getReplyTo();
-		if (replyToString == null) {
-			replyTo = null;
-		}
-		else if (replyToString.startsWith(AMQ_RABBITMQ_REPLY_TO)) {
-			replyTo = new Address("", replyToString);
-		}
-		else {
-			replyTo = new Address(replyToString);
-		}
-		return replyTo;
+		return request.getMessageProperties().getReplyToAddress();
 	}
 
 }

spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -25,6 +25,7 @@
  * @author Mark Pollack
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  */
 public class AddressTests {
 
@@ -83,4 +84,18 @@ public void testEmpty() {
 		assertEquals("/", address.toString());
 	}
 
+	@Test
+	public void testDirectReplyTo() {
+		String replyTo = Address.AMQ_RABBITMQ_REPLY_TO + ".ab/cd/ef";
+		MessageProperties props = new MessageProperties();
+		props.setReplyTo(replyTo);
+		Message message = new Message("foo".getBytes(), props);
+		Address address = AddressUtils.decodeReplyToAddress(message);
+		assertEquals("", address.getExchangeName());
+		assertEquals(replyTo, address.getRoutingKey());
+		address = props.getReplyToAddress();
+		assertEquals("", address.getExchangeName());
+		assertEquals(replyTo, address.getRoutingKey());
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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
@@ -491,13 +491,13 @@ public Collection<CorrelationData> getUnconfirmed(long age) {
 
 	private void evaluateFastReplyTo() {
 		this.usingFastReplyTo = false;
-		if (this.replyQueue == null || AddressUtils.AMQ_RABBITMQ_REPLY_TO.equals(this.replyQueue.getName())) {
+		if (this.replyQueue == null || Address.AMQ_RABBITMQ_REPLY_TO.equals(this.replyQueue.getName())) {
 			try {
 				execute(new ChannelCallback<Void>() {
 
 					@Override
 					public Void doInRabbit(Channel channel) throws Exception {
-						channel.queueDeclarePassive(AddressUtils.AMQ_RABBITMQ_REPLY_TO);
+						channel.queueDeclarePassive(Address.AMQ_RABBITMQ_REPLY_TO);
 						return null;
 					}
 				});
@@ -910,7 +910,7 @@ public Message doInRabbit(Channel channel) throws Exception {
 						"Send-and-receive methods can only be used if the Message does not already have a replyTo property.");
 				String replyTo;
 				if (RabbitTemplate.this.usingFastReplyTo) {
-					replyTo = AddressUtils.AMQ_RABBITMQ_REPLY_TO;
+					replyTo = Address.AMQ_RABBITMQ_REPLY_TO;
 				}
 				else {
 					DeclareOk queueDeclaration = channel.queueDeclare();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 the original author or authors.
+ * Copyright 2010-2015 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
@@ -59,7 +59,6 @@
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Address;
-import org.springframework.amqp.core.AddressUtils;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
@@ -1074,7 +1073,7 @@ public void testSendAndReceiveFastImplicit() {
 
 	@Test
 	public void testSendAndReceiveFastExplicit() {
-		this.template.setReplyQueue(new Queue(AddressUtils.AMQ_RABBITMQ_REPLY_TO));
+		this.template.setReplyQueue(new Queue(Address.AMQ_RABBITMQ_REPLY_TO));
 		sendAndReceiveFastGuts();
 	}
 
@@ -1084,7 +1083,7 @@ private void sendAndReceiveFastGuts() {
 
 				@Override
 				public Void doInRabbit(Channel channel) throws Exception {
-					channel.queueDeclarePassive(AddressUtils.AMQ_RABBITMQ_REPLY_TO);
+					channel.queueDeclarePassive(Address.AMQ_RABBITMQ_REPLY_TO);
 					return null;
 				}
 			});
@@ -1109,7 +1108,7 @@ public Message handleMessage(Message message) {
 			Object result = this.template.convertSendAndReceive("foo");
 			container.stop();
 			assertEquals("FOO", result);
-			assertThat(replyToWas.get(), startsWith(AddressUtils.AMQ_RABBITMQ_REPLY_TO));
+			assertThat(replyToWas.get(), startsWith(Address.AMQ_RABBITMQ_REPLY_TO));
 		}
 		catch (Exception e) {
 			assertThat(e.getCause().getCause().getMessage(), containsString("404"));