From b6b61c4cfe2ce6b6ee8c1639bd11eaa3662f3cba Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 10 Mar 2025 10:29:53 -0400 Subject: [PATCH] Improve `RabbitAmqpUtils.toAmqpMessage()` The `RabbitAmqpUtils.toAmqpMessage()` utility is used on the publisher side, so, it is natural to treat such a message as a reply. Therefore, the `correlationId` is set to `messageId` of `correlationId` is not present. The `replyTo` of the Spring message is set into `to` of the AMQP message * Mention in the `Address` JavaDocs that just `routingKey` can be treated differently by clients * Fix error message in the `RabbitAmqpMessageListenerAdapter` --- .../org/springframework/amqp/core/Address.java | 15 +++++++++------ .../amqp/rabbitmq/client/RabbitAmqpUtils.java | 15 +++++++++++---- .../RabbitAmqpMessageListenerAdapter.java | 2 +- 3 files changed, 21 insertions(+), 11 deletions(-) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java b/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java index 564c752d..36a940e5 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java @@ -23,8 +23,8 @@ 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. + * 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 * follow these conventions, which can be easily summarised as: * @@ -33,7 +33,10 @@ import org.springframework.util.StringUtils; * * * Here we also the exchange name to default to empty - * (so just a routing key will work if you know the queue name). + * (so just a routing key will work as a queue name). + *

+ * For AMQP 1.0, only routing key is treated as target destination. + * * * @author Mark Pollack * @author Mark Fisher @@ -58,11 +61,11 @@ public class Address { /** * Create an Address instance from a structured String with the form - * * 

 	 * (exchange)/(routingKey)
 	 *
* . + * If exchange is parsed to empty string, then routing key is treated as a queue name. * @param address a structured string. */ public Address(String address) { @@ -120,9 +123,9 @@ public class Address { @Override public int hashCode() { - int result = this.exchangeName != null ? this.exchangeName.hashCode() : 0; + int result = this.exchangeName.hashCode(); int prime = 31; // NOSONAR magic # - result = prime * result + (this.routingKey != null ? this.routingKey.hashCode() : 0); + result = prime * result + this.routingKey.hashCode(); return result; } diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java index 44e1df59..c10ebebd 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java @@ -19,6 +19,7 @@ package org.springframework.amqp.rabbitmq.client; import java.nio.charset.StandardCharsets; import java.util.Date; import java.util.Map; +import java.util.Objects; import java.util.UUID; import com.rabbitmq.client.amqp.Consumer; @@ -72,8 +73,12 @@ public final class RabbitAmqpUtils { } /** - * Convert {@link com.rabbitmq.client.amqp.Message} into {@link Message}. - * @param amqpMessage the {@link com.rabbitmq.client.amqp.Message} convert from. + * Convert {@link Message} into {@link com.rabbitmq.client.amqp.Message}. + * The {@link MessageProperties#getReplyTo()} is set into {@link com.rabbitmq.client.amqp.Message#to(String)}. + * The {@link com.rabbitmq.client.amqp.Message#correlationId(long)} is set to + * {@link MessageProperties#getCorrelationId()} if present, or to {@link MessageProperties#getMessageId()}. + * @param message the {@link Message} convert from. + * @param amqpMessage the {@link com.rabbitmq.client.amqp.Message} convert into. */ public static void toAmqpMessage(Message message, com.rabbitmq.client.amqp.Message amqpMessage) { MessageProperties messageProperties = message.getMessageProperties(); @@ -83,9 +88,11 @@ public final class RabbitAmqpUtils { .contentEncoding(messageProperties.getContentEncoding()) .contentType(messageProperties.getContentType()) .messageId(messageProperties.getMessageId()) - .correlationId(messageProperties.getCorrelationId()) + .correlationId( + Objects.requireNonNullElse( + messageProperties.getCorrelationId(), messageProperties.getMessageId())) .priority(messageProperties.getPriority().byteValue()) - .replyTo(messageProperties.getReplyTo()); + .to(messageProperties.getReplyTo()); Map headers = messageProperties.getHeaders(); if (!headers.isEmpty()) { diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java index a94e2f88..0bd49bf6 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java @@ -120,7 +120,7 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd InvocationResult result = getHandlerAdapter() .invoke(converted, amqpAcknowledgment); if (result.getReturnValue() != null) { - logger.warn("Replies are not currently supported with RabbitMQ AMQP 1.0 listeners"); + logger.warn("Replies for batches are not currently supported with RabbitMQ AMQP 1.0 listeners"); } } catch (Exception ex) {