Fix reply-to handling logic for AMQP 1.0

The `reply-to` property from request messages is parsed to the
`org.springframework.amqp.core.Address` object.
This one removes the first `/` leaving `routingKey`, based on the mentioned `reply-to`,
 with a `queues/` prefix.
The value in the `reply-to` is already encoded, so we don't need to go extra parsing logic
in the `RabbitAmqpTemplate.toAmqpMessage()` in regards
`com.rabbitmq.client.amqp.Message.MessageAddressBuilder`

* Fix `RabbitAmqpTemplate.toAmqpMessage()` to check for `queues/` prefix before going down to
the `com.rabbitmq.client.amqp.Message.MessageAddressBuilder` logic.
Instead, use the `queue` value as is in the  `com.rabbitmq.client.amqp.Message.to()` property
adding required `/` at the beginning of the value
* Revert `RabbitAmqpMessageListenerAdapter.sendResponse()` logic to `this.rabbitAmqpTemplate.send()`
without modifying `replyToRoutingKey` value
* Fix typos in the `Address` Javadocs

Author: artembilan

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

@@ -25,7 +25,7 @@
 /**
  * Represents an address for publication of an AMQP message. The AMQP 0.9
  * specification has an unstructured string that is used as a "reply to" address.
- * There are however conventions in use and this class makes it easier to
+ * 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">
@@ -35,7 +35,7 @@
  * Here we also the exchange name to default to empty
  * (so just a routing key will work as a queue name).
  * <p>
- * For AMQP 1.0, only routing key is treated as target destination.
+ * For AMQP 1.0, only a routing key is treated as a target destination.
  *
  *
  * @author Mark Pollack
@@ -65,7 +65,8 @@ public class Address {
 	 * (exchange)/(routingKey)
 	 * </pre>
 	 * .
-	 * If exchange is parsed to empty string, then routing key is treated as a queue name.
+	 * If exchange is parsed to an empty string, then a routing key is treated as a queue name.
+	 * The {@link #AMQ_RABBITMQ_REPLY_TO} matching address is treated as a routing key.
 	 * @param address a structured string.
 	 */
 	public Address(String address) {

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java

@@ -663,13 +663,26 @@ private static com.rabbitmq.client.amqp.Message toAmqpMessage(@Nullable String e
 				amqpMessageSupplier.get()
 						.toAddress();
 
-		JavaUtils.INSTANCE
-				.acceptIfNotNull(exchange, address::exchange)
-				.acceptIfNotNull(routingKey, address::key)
-				.acceptIfNotNull(queue, address::queue);
+		/*
+		 * The 'reply-to' property in the request comes already encoded and with the '/queues/' prefix.
+		 * However, the 'org.springframework.amqp.core.Address' removes the first '/' on parsing.
+		 * Use this value as is, skipping 'com.rabbitmq.client.amqp.Message.MessageAddressBuilder' logic.
+		 */
+		boolean replyToProperty = queue != null && queue.startsWith("queues/");
+
+		if (!replyToProperty) {
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(exchange, address::exchange)
+					.acceptIfNotNull(routingKey, address::key)
+					.acceptIfNotNull(queue, address::queue);
+		}
 
 		com.rabbitmq.client.amqp.Message amqpMessage = address.message();
 
+		if (replyToProperty) {
+			amqpMessage.to('/' + queue);
+		}
+
 		RabbitAmqpUtils.toAmqpMessage(message, amqpMessage);
 
 		return amqpMessage;

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java

@@ -24,14 +24,12 @@
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.amqp.Consumer;
-import com.rabbitmq.client.amqp.Publisher;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.core.Address;
 import org.springframework.amqp.core.AmqpAcknowledgment;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
-import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
 import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
@@ -157,19 +155,7 @@ protected void sendResponse(@Nullable Channel channel, Address replyTo, Message
 			sendFuture = this.rabbitAmqpTemplate.send(replyToExchange, replyToRoutingKey, replyMessage);
 		}
 		else {
-			Assert.hasText(replyToRoutingKey, "The 'replyTo' must be provided, in request message or in @SendTo.");
-			Publisher publisher = this.rabbitAmqpTemplate.getPublisher();
-			com.rabbitmq.client.amqp.Message amqpMessage = publisher.message();
-			RabbitAmqpUtils.toAmqpMessage(replyMessage, amqpMessage);
-			amqpMessage.to("/queues/" + replyToRoutingKey.replaceFirst("queues/", ""));
-			sendFuture = new CompletableFuture<>();
-			publisher.publish(amqpMessage, (context) -> {
-				switch (context.status()) {
-					case ACCEPTED -> sendFuture.complete(true);
-					case REJECTED, RELEASED -> sendFuture.completeExceptionally(
-							new AmqpNackReceivedException("The message was rejected", messageIn));
-				}
-			});
+			sendFuture = this.rabbitAmqpTemplate.send(replyToRoutingKey, replyMessage);
 		}
 
 		sendFuture.join();