Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2590: RepubMRecoverWithConfirms: SpEL-based ctor

Browse Source 
Fixes: #2590

Expose the SpEL-based `errorExchange` & `errorRoutingKey` on `RepublishMessageRecovererWithConfirms`
 like it was done in the `RepublishMessageRecoverer`.

* Add `RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, @Nullable Expression errorExchange, @Nullable Expression errorRoutingKey, ConfirmType confirmType)`
* Fix typos in `RepublishMessageRecovererWithConfirms` Javadocs
* Modify `RepublishMessageRecovererWithConfirmsIntegrationTests.testCorrelatedWithNack()`
to use SpEL expressions instead of plain strings
This commit is contained in:
Artem Bilan
2024-02-15 12:07:18 -05:00
parent 95d292a297
commit 2ad69c1820
2 changed files with 48 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2024 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.
@@ -28,12 +28,15 @@ import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.expression.Expression;
import org.springframework.lang.Nullable;
/**
* A {@link RepublishMessageRecoverer} supporting publisher confirms and returns.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.3.3
*
*/
@@ -48,20 +51,20 @@ public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecov
private long confirmTimeout = DEFAULT_TIMEOUT;
/**
* Use the supplied template to publish the messsage with the provided confirm type.
* Use the supplied template to publish the message with the provided confirm type.
* The template and its connection factory must be suitably configured to support the
* confirm type.
* {@code confirm} type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
*/
public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, ConfirmType confirmType) {
this(errorTemplate, null, null, confirmType);
this(errorTemplate, (Expression) null, null, confirmType);
}
/**
* Use the supplied template to publish the messsage with the provided confirm type to
* Use the supplied template to publish the message with the provided confirm type to
* the provided exchange with the default routing key. The template and its connection
* factory must be suitably configured to support the confirm type.
* factory must be suitably configured to support the {@code confirm} type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
* @param errorExchange the exchange.
@@ -73,9 +76,9 @@ public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecov
}
/**
* Use the supplied template to publish the messsage with the provided confirm type to
* Use the supplied template to publish the message with the provided confirm type to
* the provided exchange with the provided routing key. The template and its
* connection factory must be suitably configured to support the confirm type.
* connection factory must be suitably configured to support the {@code confirm} type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
* @param errorExchange the exchange.
@@ -90,7 +93,25 @@ public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecov
}
/**
* Set the confirm timeout; default 10 seconds.
* Use the supplied template to publish the message with the provided confirm type to
* the provided exchange with the provided routing key. The template and its
* connection factory must be suitably configured to support the {@code confirm} type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
* @param errorExchange the exchange.
* @param errorRoutingKey the routing key.
* @since 3.1.2
*/
public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, @Nullable Expression errorExchange,
@Nullable Expression errorRoutingKey, ConfirmType confirmType) {
super(errorTemplate, errorExchange, errorRoutingKey);
this.template = errorTemplate;
this.confirmType = confirmType;
}
/**
* Set the {@code confirm} timeout; default 10 seconds.
* @param confirmTimeout the timeout.
*/
public void setConfirmTimeout(long confirmTimeout) {
@@ -98,8 +119,7 @@ public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecov
}
@Override
protected void doSend(@Nullable
String exchange, String routingKey, Message message) {
protected void doSend(@Nullable String exchange, String routingKey, Message message) {
if (ConfirmType.CORRELATED.equals(this.confirmType)) {
doSendCorrelated(exchange, routingKey, message);
}

23
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirmsIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2023 the original author or authors.
* Copyright 2018-2024 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.
@@ -35,6 +35,8 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
/**
* @author Gary Russell
@@ -106,12 +108,21 @@ class RepublishMessageRecovererWithConfirmsIntegrationTests {
.maxLength(1L)
.overflow(Overflow.rejectPublish)
.build();
admin.deleteQueue(queue.getName());
admin.declareQueue(queue);
RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "",
queue.getName(), ConfirmType.CORRELATED);
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException());
assertThatExceptionOfType(AmqpNackReceivedException.class).isThrownBy(() ->
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException()));
RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template,
new LiteralExpression(""),
new SpelExpressionParser().parseExpression("messageProperties.headers[queueName]"),
ConfirmType.CORRELATED);
Message message = MessageBuilder.withBody("foo".getBytes()).setHeader("queueName", queue.getName()).build();
recoverer.recover(message, new RuntimeException());
assertThatExceptionOfType(AmqpNackReceivedException.class)
.isThrownBy(() -> recoverer.recover(message, new RuntimeException()));
admin.deleteQueue(queue.getName());
ccf.destroy();
}