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 d64eb669..b8ed61f4 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -20,6 +20,7 @@ import java.util.UUID; import org.springframework.amqp.core.Correlation; import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.ReturnedMessage; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.concurrent.SettableListenableFuture; @@ -29,7 +30,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} methods that include one of * these as a parameter; when the publisher confirm is received, the CorrelationData is * returned with the ack/nack. When returns are also enabled, the - * {@link #setReturnedMessage(Message) returnedMessage} property will be populated when a + * {@link #setReturned(ReturnedMessage) returned} 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. If no id is provided it * will automatically set to a unique value. @@ -44,7 +45,7 @@ public class CorrelationData implements Correlation { private volatile String id; - private volatile Message returnedMessage; + private volatile ReturnedMessage returnedMessage; /** * Construct an instance with a null Id. @@ -56,7 +57,7 @@ public class CorrelationData implements Correlation { /** * Construct an instance with the supplied id. Must be unique if returns are enabled - * to allow population of the {@link #setReturnedMessage(Message) returnedMessage}. + * to allow population of the {@link #setReturned(ReturnedMessage) returned} message. * @param id the id. */ public CorrelationData(String id) { @@ -98,21 +99,46 @@ public class CorrelationData implements Correlation { * Return a returned message, if any; requires a unique * {@link #CorrelationData(String) id}. Guaranteed to be populated before the future * is set. + * @deprecated in favor of {@link #getReturned()}. * @return the message or null. * @since 2.1 */ + @Deprecated @Nullable public Message getReturnedMessage() { - return this.returnedMessage; + return this.returnedMessage.getMessage(); } /** * Set a returned message for this correlation data. * @param returnedMessage the returned message. + * @deprecated in favor of {@link #setReturned(ReturnedMessage)}. * @since 1.7.13 */ + @Deprecated public void setReturnedMessage(Message returnedMessage) { - this.returnedMessage = returnedMessage; + this.returnedMessage = new ReturnedMessage(returnedMessage, 0, "not available", "not available", + "not available"); + } + + /** + * Get the returned message and metadata, if any. Guaranteed to be populated before + * the future is set. + * @return the {@link ReturnedMessage}. + * @since 2.3.3 + */ + @Nullable + public ReturnedMessage getReturned() { + return this.returnedMessage; + } + + /** + * Set the returned message and metadata. + * @param returned the {@link ReturnedMessage}. + * @since 2.3.3 + */ + public void setReturned(ReturnedMessage returned) { + this.returnedMessage = returned; } @Override diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java index 1871896f..de51471e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -40,6 +40,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; @@ -1101,7 +1102,9 @@ public class PublisherCallbackChannelImpl new Envelope(0L, false, returned.getExchange(), returned.getRoutingKey()), StandardCharsets.UTF_8.name()); if (confirm.getCorrelationData() != null) { - confirm.getCorrelationData().setReturnedMessage(new Message(returned.getBody(), messageProperties)); // NOSONAR never null + confirm.getCorrelationData().setReturned(new ReturnedMessage(// NOSONAR never null + new Message(returned.getBody(), messageProperties), returned.getReplyCode(), + returned.getReplyText(), returned.getExchange(), returned.getRoutingKey())); } } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/AmqpNackReceivedException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/AmqpNackReceivedException.java new file mode 100644 index 00000000..fb8456e0 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/AmqpNackReceivedException.java @@ -0,0 +1,54 @@ +/* + * Copyright 2021 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 + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.core; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.Message; + +/** + * An exception thrown when a negative acknowledgement received after publishing a + * message. + * + * @author Gary Russell + * @since 2.3.3 + * + */ +public class AmqpNackReceivedException extends AmqpException { + + private static final long serialVersionUID = 1L; + + private final Message failedMessage; + + /** + * Create an instance with the provided message and failed message. + * @param message the message. + * @param failedMessage the failed message. + */ + public AmqpNackReceivedException(String message, Message failedMessage) { + super(message); + this.failedMessage = failedMessage; + } + + /** + * Return the failed message. + * @return the message. + */ + public Message getFailedMessage() { + return this.failedMessage; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java index 2bf670e3..271b5c2d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2021 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. @@ -29,6 +29,7 @@ import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -177,7 +178,7 @@ public class RepublishMessageRecoverer implements MessageRecoverer { if (null != this.errorExchangeName) { String routingKey = this.errorRoutingKey != null ? this.errorRoutingKey : this.prefixedOriginalRoutingKey(message); - this.errorTemplate.send(this.errorExchangeName, routingKey, message); + doSend(this.errorExchangeName, routingKey, message); if (this.logger.isWarnEnabled()) { this.logger.warn("Republishing failed message to exchange '" + this.errorExchangeName + "' with routing key " + routingKey); @@ -185,7 +186,7 @@ public class RepublishMessageRecoverer implements MessageRecoverer { } else { final String routingKey = this.prefixedOriginalRoutingKey(message); - this.errorTemplate.send(routingKey, message); + doSend(null, routingKey, message); if (this.logger.isWarnEnabled()) { this.logger.warn("Republishing failed message to the template's default exchange with routing key " + routingKey); @@ -193,6 +194,22 @@ public class RepublishMessageRecoverer implements MessageRecoverer { } } + /** + * Send the message. + * @param exchange the exchange or null to use the template's default. + * @param routingKey the routing key. + * @param message the message. + * @since 2.3.3 + */ + protected void doSend(@Nullable String exchange, String routingKey, Message message) { + if (exchange != null) { + this.errorTemplate.send(exchange, routingKey, message); + } + else { + this.errorTemplate.send(routingKey, message); + } + } + private String[] processStackTrace(Throwable cause, String exceptionMessage) { String stackTraceAsString = getStackTraceAsString(cause); if (this.maxStackTraceLength < 0) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java new file mode 100644 index 00000000..1c65f593 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java @@ -0,0 +1,153 @@ +/* + * Copyright 2021 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 + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.retry; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.springframework.amqp.core.AmqpMessageReturnedException; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; +import org.springframework.amqp.rabbit.connection.CorrelationData; +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.lang.Nullable; + +/** + * A {@link RepublishMessageRecoverer} supporting publisher confirms and returns. + * + * @author Gary Russell + * @since 2.3.3 + * + */ +public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecoverer { + + private static final long DEFAULT_TIMEOUT = 10_000; + + private final RabbitTemplate template; + + private final ConfirmType confirmType; + + private long confirmTimeout = DEFAULT_TIMEOUT; + + /** + * Use the supplied template to publish the messsage with the provided confirm type. + * The template and its connection factory must be suitably configured to support the + * confirm type. + * @param errorTemplate the template. + * @param confirmType the confirmType. + */ + public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, ConfirmType confirmType) { + this(errorTemplate, null, null, confirmType); + } + + /** + * Use the supplied template to publish the messsage 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. + * @param errorTemplate the template. + * @param confirmType the confirmType. + * @param errorExchange the exchange. + */ + public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, String errorExchange, + ConfirmType confirmType) { + + this(errorTemplate, errorExchange, null, confirmType); + } + + /** + * Use the supplied template to publish the messsage 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. + * @param errorTemplate the template. + * @param confirmType the confirmType. + * @param errorExchange the exchange. + * @param errorRoutingKey the routing key. + */ + public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, String errorExchange, + String errorRoutingKey, ConfirmType confirmType) { + + super(errorTemplate, errorExchange, errorRoutingKey); + this.template = errorTemplate; + this.confirmType = confirmType; + } + + /** + * Set the confirm timeout; default 10 seconds. + * @param confirmTimeout the timeout. + */ + public void setConfirmTimeout(long confirmTimeout) { + this.confirmTimeout = confirmTimeout; + } + + @Override + protected void doSend(@Nullable + String exchange, String routingKey, Message message) { + if (ConfirmType.CORRELATED.equals(this.confirmType)) { + doSendCorrelated(exchange, routingKey, message); + } + else { + doSendSimple(exchange, routingKey, message); + } + } + + private void doSendCorrelated(String exchange, String routingKey, Message message) { + CorrelationData cd = new CorrelationData(); + if (exchange != null) { + this.template.send(exchange, routingKey, message, cd); + } + else { + this.template.send(routingKey, message, cd); + } + try { + Confirm confirm = cd.getFuture().get(this.confirmTimeout, TimeUnit.MILLISECONDS); + if (cd.getReturned() != null) { + throw new AmqpMessageReturnedException("Message returned", cd.getReturned()); + } + if (!confirm.isAck()) { + throw new AmqpNackReceivedException("Negative acknowledgment received", message); + } + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw RabbitExceptionTranslator.convertRabbitAccessException(ex); + } + catch (ExecutionException ex) { + throw RabbitExceptionTranslator.convertRabbitAccessException(ex.getCause()); // NOSONAR (stack trace) + } + catch (TimeoutException ex) { + throw RabbitExceptionTranslator.convertRabbitAccessException(ex); + } + } + + private void doSendSimple(String exchange, String routingKey, Message message) { + this.template.invoke(sender -> { + if (exchange != null) { + sender.send(exchange, routingKey, message); + } + else { + sender.send(routingKey, message); + } + sender.waitForConfirmsOrDie(this.confirmTimeout); + return null; + }); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java index bad6a159..e48dc3b2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -66,7 +66,7 @@ public class MessagingTemplateConfirmsTests { rmt.send("messaging.confirms.unroutable", new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data))); assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); - assertThat(data.getReturnedMessage()).isNotNull(); + assertThat(data.getReturned()).isNotNull(); ccf.destroy(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index d827284d..550eee4e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -855,7 +855,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { this.templateWithConfirmsAndReturnsEnabled.convertAndSend("", "NO_QUEUE_HERE", "foo", cd4); assertThat(cd4.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); assertThat(callbackLatch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(cd4.getReturnedMessage()).isNotNull(); + assertThat(cd4.getReturned()).isNotNull(); assertThat(resent.get()).isTrue(); assertThat(callbackThreadName.get()).startsWith("spring-rabbit-deferred-pool"); admin.deleteQueue(queue.getName()); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java index 36cbc991..8d46fb24 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -125,7 +125,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 { corr = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("", "bad route", "foo", corr); assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); - assertThat(corr.getReturnedMessage()).isNotNull(); + assertThat(corr.getReturned()).isNotNull(); } @Test diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java index 7b224c49..8e7fcef7 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2021 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. @@ -103,7 +103,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 { assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(cacheCount.get()).isEqualTo(1); assertThat(returnCalledFirst.get()).isTrue(); - assertThat(correlationData.getReturnedMessage()).isNotNull(); + assertThat(correlationData.getReturned()).isNotNull(); cf.destroy(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirmsIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirmsIntegrationTests.java new file mode 100644 index 00000000..dae6bf2e --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirmsIntegrationTests.java @@ -0,0 +1,119 @@ +/* + * Copyright 2018-2019 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 + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.retry; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.fail; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.AmqpMessageReturnedException; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageBuilder; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.QueueBuilder; +import org.springframework.amqp.core.QueueBuilder.Overflow; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; +import org.springframework.amqp.rabbit.core.AmqpNackReceivedException; +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; + +/** + * @author Gary Russell + * @since 2.0.5 + * + */ +@RabbitAvailable(queues = RepublishMessageRecovererWithConfirmsIntegrationTests.QUEUE) +class RepublishMessageRecovererWithConfirmsIntegrationTests { + + static final String QUEUE = "RepublishMessageRecovererWithConfirmsIntegrationTests"; + + @Test + void testSimple() { + CachingConnectionFactory ccf = new CachingConnectionFactory( + RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + ccf.setPublisherConfirmType(ConfirmType.SIMPLE); + RabbitTemplate template = new RabbitTemplate(ccf); + RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "", + QUEUE, ConfirmType.SIMPLE); + recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException()); + Message received = template.receive(QUEUE, 10_000); + assertThat(received.getBody()).isEqualTo("foo".getBytes()); + ccf.destroy(); + } + + @Test + void testCorrelated() { + CachingConnectionFactory ccf = new CachingConnectionFactory( + RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); + RabbitTemplate template = new RabbitTemplate(ccf); + RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "", + QUEUE, ConfirmType.CORRELATED); + recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException()); + Message received = template.receive(QUEUE, 10_000); + assertThat(received.getBody()).isEqualTo("foo".getBytes()); + ccf.destroy(); + } + + @Test + void testCorrelatedNotRoutable() { + CachingConnectionFactory ccf = new CachingConnectionFactory( + RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); + ccf.setPublisherReturns(true); + RabbitTemplate template = new RabbitTemplate(ccf); + template.setMandatory(true); + RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "", + "bad.route", ConfirmType.CORRELATED); + try { + recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException()); + fail("Expected exception"); + } + catch (AmqpMessageReturnedException ex) { + assertThat(ex.getReturnedMessage().getBody()).isEqualTo("foo".getBytes()); + assertThat(ex.getReplyText()).isEqualTo("NO_ROUTE"); + } + ccf.destroy(); + } + + @Test + void testCorrelatedWithNack() { + CachingConnectionFactory ccf = new CachingConnectionFactory( + RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + ccf.setPublisherConfirmType(ConfirmType.CORRELATED); + RabbitTemplate template = new RabbitTemplate(ccf); + RabbitAdmin admin = new RabbitAdmin(ccf); + Queue queue = QueueBuilder.durable(QUEUE + ".nack") + .maxLength(1) + .overflow(Overflow.rejectPublish) + .build(); + 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())); + admin.deleteQueue(queue.getName()); + ccf.destroy(); + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 69decc0f..41116aa2 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -6263,6 +6263,14 @@ Starting with versions 2.1.13, 2.2.3, the exception message is included in this Whenever a truncation of any kind occurs, the original exception will be logged to retain the complete information. +Starting with version 2.3.3, a new subclass `RepublishMessageRecovererWithConfirms` is provided; this supports both styles of publisher confirms and will wait for the confirmation before returning (or throw an exception if not confirmed or the message is returned). + +If the confirm type is `CORRELATED`, the subclass will also detect if a message is returned and throw an `AmqpMessageReturnedException`; if the publication is negatively acknowledged, it will throw an `AmqpNackReceivedException`. + +If the confirm type is `SIMPLE`, the subclass will invoke the `waitForConfirmsOrDie` method on the channel. + +See <> for more information about confirms and returns. + Starting with version 2.1, an `ImmediateRequeueMessageRecoverer` is added to throw an `ImmediateRequeueAmqpException`, which notifies a listener container to requeue the current failed message. ===== Exception Classification for Spring Retry diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index ec6a56a7..40099e71 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -52,3 +52,8 @@ See the IMPORTANT note in <> for more information. ==== Multiple Broker Support Improvements See <> for more information. + +==== RepublishMessageRecoverer Changes + +A new subclass of this recoverer is not provided that supports publisher confirms. +See <> for more information.