From ea546c52d0c248a7ee3b5ce1102117101da2681b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 30 Jul 2020 11:00:28 -0400 Subject: [PATCH] Fix compatibility with latest Spring AMQP (#3352) * Fix compatibility with latest Spring AMQP * * Fix typo in the `amqp.adoc` --- .../AbstractAmqpOutboundEndpoint.java | 40 ++++++++++++++----- .../amqp/outbound/AmqpOutboundEndpoint.java | 13 +++--- .../outbound/AsyncAmqpOutboundGateway.java | 6 ++- src/reference/asciidoc/amqp.adoc | 2 +- 4 files changed, 40 insertions(+), 21 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index 38c196588d..be43c1f76f 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -24,6 +24,7 @@ import java.util.UUID; import java.util.concurrent.ScheduledFuture; import org.springframework.amqp.core.MessageDeliveryMode; +import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.CorrelationData; @@ -602,25 +603,44 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin : getMessageBuilderFactory().withPayload(replyObject); } + /** + * Build Spring message object based on the provided returned AMQP message info. + * @param message the returned AMQP message + * @param replyCode the returned message reason code + * @param replyText the returned message reason text + * @param exchange the exchange the message returned from + * @param returnedRoutingKey the routing key for returned message + * @param converter the converter to deserialize body of the returned AMQP message + * @return the Spring message which represents a returned AMQP message + * @deprecated since 5.4 in favor of {@link #buildReturnedMessage(ReturnedMessage, MessageConverter)} + */ + @Deprecated protected Message buildReturnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String returnedRoutingKey, MessageConverter converter) { - Object returnedObject = converter.fromMessage(message); + return buildReturnedMessage(new ReturnedMessage(message, replyCode, replyText, exchange, returnedRoutingKey), + converter); + } + + protected Message buildReturnedMessage(ReturnedMessage returnedMessage, MessageConverter converter) { + org.springframework.amqp.core.Message amqpMessage = returnedMessage.getMessage(); + Object returnedObject = converter.fromMessage(amqpMessage); AbstractIntegrationMessageBuilder builder = prepareMessageBuilder(returnedObject); - Map headers = getHeaderMapper().toHeadersFromReply(message.getMessageProperties()); + Map headers = getHeaderMapper().toHeadersFromReply(amqpMessage.getMessageProperties()); if (this.errorMessageStrategy == null) { builder.copyHeadersIfAbsent(headers) - .setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode) - .setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText) - .setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange) - .setHeader(AmqpHeaders.RETURN_ROUTING_KEY, returnedRoutingKey); + .setHeader(AmqpHeaders.RETURN_REPLY_CODE, returnedMessage.getReplyCode()) + .setHeader(AmqpHeaders.RETURN_REPLY_TEXT, returnedMessage.getReplyText()) + .setHeader(AmqpHeaders.RETURN_EXCHANGE, returnedMessage.getExchange()) + .setHeader(AmqpHeaders.RETURN_ROUTING_KEY, returnedMessage.getRoutingKey()); } - Message returnedMessage = builder.build(); + Message message = builder.build(); if (this.errorMessageStrategy != null) { - returnedMessage = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException( - returnedMessage, message, replyCode, replyText, exchange, returnedRoutingKey), null); + message = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException( + message, amqpMessage, returnedMessage.getReplyCode(), returnedMessage.getReplyText(), + returnedMessage.getExchange(), returnedMessage.getRoutingKey()), null); } - return returnedMessage; + return message; } protected void handleConfirm(CorrelationData correlationData, boolean ack, String cause) { diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index 6c1bda02cb..73c5eafc85 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -23,11 +23,11 @@ import java.util.concurrent.TimeoutException; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; -import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.Lifecycle; import org.springframework.integration.IntegrationPatternType; @@ -48,7 +48,7 @@ import org.springframework.util.Assert; * @since 2.1 */ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint - implements ConfirmCallback, ReturnCallback { + implements ConfirmCallback, RabbitTemplate.ReturnsCallback { private static final Duration DEFAULT_CONFIRM_TIMEOUT = Duration.ofSeconds(5); @@ -141,7 +141,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint if (getReturnChannel() != null) { Assert.notNull(this.rabbitTemplate, "RabbitTemplate implementation is required for publisher confirms"); - this.rabbitTemplate.setReturnCallback(this); + this.rabbitTemplate.setReturnsCallback(this); } Duration confirmTimeout = getConfirmTimeout(); if (confirmTimeout != null) { @@ -267,13 +267,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint } @Override - public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, - String exchange, String routingKey) { - + public void returnedMessage(ReturnedMessage returnedMessage) { // no need for null check; we asserted we have a RabbitTemplate in doInit() MessageConverter converter = this.rabbitTemplate.getMessageConverter(); - Message returned = buildReturnedMessage(message, replyCode, replyText, exchange, - routingKey, converter); + Message returned = buildReturnedMessage(returnedMessage, converter); getReturnChannel().send(returned); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index d32c109a05..f5a99d5e84 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -18,6 +18,7 @@ package org.springframework.integration.amqp.outbound; import org.springframework.amqp.core.AmqpMessageReturnedException; import org.springframework.amqp.core.AmqpReplyTimeoutException; +import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CorrelationData; @@ -147,8 +148,9 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { else { AmqpMessageReturnedException amre = (AmqpMessageReturnedException) ex; Message returnedMessage = buildReturnedMessage( - amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(), amre.getExchange(), - amre.getRoutingKey(), AsyncAmqpOutboundGateway.this.messageConverter); + new ReturnedMessage(amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(), + amre.getExchange(), amre.getRoutingKey()), + AsyncAmqpOutboundGateway.this.messageConverter); sendOutput(returnedMessage, getReturnChannel(), true); } } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index cf0cea9aca..6913ddfc15 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -309,7 +309,7 @@ The following example shows how to configure an `AmqpMessageSource` with Java co ---- @Bean public AmqpMessageSource source(ConnectionFactory connectionFactory) { - return new AmpqpMessageSource(connectionFactory, "someQueue"); + return new AmqpMessageSource(connectionFactory, "someQueue"); } ---- ====