From 9868649e8cd7db158bab5eb10db6efd0b828a804 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 28 Aug 2020 17:10:35 -0400 Subject: [PATCH] Require an ID in CorrelationData Required for proper coordination of returns and confirms. It is mentioned in the javadocs, but easy to miss. --- .../amqp/rabbit/AsyncRabbitTemplate.java | 20 ++++++++++--------- .../rabbit/connection/CorrelationData.java | 19 ++++++++++++------ src/reference/asciidoc/amqp.adoc | 4 ++-- src/reference/asciidoc/whats-new.adoc | 3 +++ 4 files changed, 29 insertions(+), 17 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java index d4ab45cd..e9f7bc9c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java @@ -374,7 +374,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa @Override public RabbitMessageFuture sendAndReceive(String exchange, String routingKey, Message message) { - String correlationId = getOrSetCorrelationIdAndSetReplyTo(message); + String correlationId = getOrSetCorrelationIdAndSetReplyTo(message, null); RabbitMessageFuture future = new RabbitMessageFuture(correlationId, message); CorrelationData correlationData = null; if (this.enableConfirms) { @@ -640,13 +640,15 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa } } - private String getOrSetCorrelationIdAndSetReplyTo(Message message) { + private String getOrSetCorrelationIdAndSetReplyTo(Message message, + @Nullable AsyncCorrelationData correlationData) { + String correlationId; MessageProperties messageProperties = message.getMessageProperties(); Assert.notNull(messageProperties, "the message properties cannot be null"); String currentCorrelationId = messageProperties.getCorrelationId(); if (!StringUtils.hasText(currentCorrelationId)) { - correlationId = UUID.randomUUID().toString(); + correlationId = correlationData != null ? correlationData.getId() : UUID.randomUUID().toString(); messageProperties.setCorrelationId(correlationId); Assert.isNull(messageProperties.getReplyTo(), "'replyTo' property must be null"); } @@ -816,9 +818,9 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa if (correlationData.userPostProcessor != null) { messageToSend = correlationData.userPostProcessor.postProcessMessage(message); } - String correlationId = getOrSetCorrelationIdAndSetReplyTo(messageToSend); + String correlationId = getOrSetCorrelationIdAndSetReplyTo(messageToSend, correlationData); correlationData.future = new RabbitConverterFuture(correlationId, message); - if (correlationData.enableConfirms && correlationData.getId() == null) { + if (correlationData.enableConfirms) { correlationData.setId(correlationId); correlationData.future.setConfirm(new SettableListenableFuture<>()); } @@ -831,13 +833,13 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa private static class AsyncCorrelationData extends CorrelationData { - private final MessagePostProcessor userPostProcessor; + final MessagePostProcessor userPostProcessor; // NOSONAR - private final ParameterizedTypeReference returnType; + final ParameterizedTypeReference returnType; // NOSONAR - private final boolean enableConfirms; + final boolean enableConfirms; // NOSONAR - private volatile RabbitConverterFuture future; + volatile RabbitConverterFuture future; // NOSONAR AsyncCorrelationData(MessagePostProcessor userPostProcessor, ParameterizedTypeReference returnType, boolean enableConfirms) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java index 2d2fd232..d64eb669 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java @@ -16,9 +16,12 @@ package org.springframework.amqp.rabbit.connection; +import java.util.UUID; + import org.springframework.amqp.core.Correlation; import org.springframework.amqp.core.Message; import org.springframework.lang.Nullable; +import org.springframework.util.Assert; import org.springframework.util.concurrent.SettableListenableFuture; /** @@ -28,7 +31,8 @@ import org.springframework.util.concurrent.SettableListenableFuture; * returned with the ack/nack. When returns are also enabled, the * {@link #setReturnedMessage(Message) returnedMessage} property will be populated when a * message can't be delivered - the return always arrives before the confirmation. In this - * case the {@code #id} property must be set to a unique value. + * case the {@code #id} property must be set to a unique value. If no id is provided it + * will automatically set to a unique value. * * @author Gary Russell * @since 1.0.1 @@ -38,7 +42,6 @@ public class CorrelationData implements Correlation { private final SettableListenableFuture future = new SettableListenableFuture<>(); - @Nullable private volatile String id; private volatile Message returnedMessage; @@ -48,6 +51,7 @@ public class CorrelationData implements Correlation { * @since 1.6.7 */ public CorrelationData() { + this.id = UUID.randomUUID().toString(); } /** @@ -56,10 +60,14 @@ public class CorrelationData implements Correlation { * @param id the id. */ public CorrelationData(String id) { + Assert.notNull(id, "'id' cannot be null and must be unique"); this.id = id; } - @Nullable + /** + * Return the id. + * @return the id. + */ public String getId() { return this.id; } @@ -67,14 +75,13 @@ public class CorrelationData implements Correlation { /** * Set the correlation id. Generally, the correlation id shouldn't be changed. * One use case, however, is when it needs to be set in a - * {@link org.springframework.amqp.core.MessagePostProcessor} after a - * {@link CorrelationData} with a 'null' correlation id has been passed into a - * {@link org.springframework.amqp.rabbit.core.RabbitTemplate}. + * {@link org.springframework.amqp.core.MessagePostProcessor}. * * @param id the id. * @since 1.6 */ public void setId(String id) { + Assert.notNull(id, "'id' cannot be null and must be unique"); this.id = id; } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 40a7968c..48b41ab1 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1192,8 +1192,8 @@ The `Confirm` object is a simple bean with 2 properties: `ack` and `reason` (for The reason is not populated for broker-generated `nack` instances. It is populated for `nack` instances generated by the framework (for example, closing the connection while `ack` instances are outstanding). -In addition, when both confirms and returns are enabled, the `CorrelationData` is populated with the returned message. -It is guaranteed that this occurs before the future is set with the `ack`. +In addition, when both confirms and returns are enabled, the `CorrelationData` is populated with the returned message, as long as the `CorrelationData` has a unique `id`; this is always the case, by default, starting with version 2.3. +It is guaranteed that the return message is set before the future is set with the `ack`. See also <> for a simpler mechanism for waiting for publisher confirms. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e3e409f8..50a7bdf7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -31,6 +31,9 @@ See <> for more information. The template's `ReturnCallback` has been refactored as `ReturnsCallback` for simpler use in lambda expressions. See <> for more information. +When using returns and correlated confirms, the `CorrelationData` now requires a unique `id` property. +See <> for more information. + ==== Listener Container Changes A new listener container property `consumeDelay` is now available; it is helpful when using the https://github.com/rabbitmq/rabbitmq-sharding[RabbitMQ Sharding Plugin].