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

GH-2797: AMQP: Add confirm-timeout

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/2797

Avoid runtime casting to `RabbitTemplate`.
This commit is contained in:
Gary Russell
2019-03-11 15:04:57 -04:00
committed by Artem Bilan
parent e13aa288b6
commit 5245effd32
13 changed files with 243 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-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.
@@ -80,6 +80,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression",

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-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.
@@ -94,6 +94,7 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
BeanDefinitionBuilder mapperBuilder = BeanDefinitionBuilder

53
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
View File

@@ -16,14 +16,18 @@
package org.springframework.integration.amqp.outbound;
import java.time.Duration;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledFuture;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
@@ -40,6 +44,7 @@ import org.springframework.integration.handler.ExpressionEvaluatingMessageProces
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -100,8 +105,12 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
private Duration confirmTimeout;
private volatile boolean running;
private volatile ScheduledFuture<?> confirmChecker;
/**
* Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers.
* Defaults to {@link DefaultAmqpHeaderMapper#outboundMapper()}.
@@ -311,6 +320,18 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
this.errorMessageStrategy = errorMessageStrategy;
}
/**
* Set a timeout after which a nack will be synthesized if no publisher confirm has
* been received within that time. Missing confirms will be checked every 50% of this
* value so the synthesized nack will be sent between 1x and 1.5x this timeout.
* @param confirmTimeout the approximate timeout.
* @since 5.2
* @see #setConfirmNackChannel(MessageChannel)
*/
public void setConfirmTimeout(long confirmTimeout) {
this.confirmTimeout = Duration.ofMillis(confirmTimeout);
}
protected final synchronized void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
@@ -381,6 +402,10 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
return this.headersMappedLast;
}
protected Duration getConfirmTimeout() {
return this.confirmTimeout;
}
@Override
protected final void doInit() {
Assert.state(this.exchangeNameExpression == null || this.exchangeName == null,
@@ -411,7 +436,8 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
}
else {
NullChannel nullChannel = extractTypeIfPossible(this.confirmAckChannel, NullChannel.class);
Assert.state((this.confirmAckChannel == null || nullChannel != null) && this.confirmAckChannelName == null,
Assert.state(
(this.confirmAckChannel == null || nullChannel != null) && this.confirmAckChannelName == null,
"A 'confirmCorrelationExpression' is required when specifying a 'confirmAckChannel'");
nullChannel = extractTypeIfPossible(this.confirmNackChannel, NullChannel.class);
Assert.state(
@@ -450,16 +476,35 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
}
}
doStart();
if (this.confirmTimeout != null && getConfirmNackChannel() != null && getRabbitTemplate() != null) {
this.confirmChecker = getTaskScheduler()
.scheduleAtFixedRate(checkUnconfirmed(), this.confirmTimeout.dividedBy(2L));
}
this.running = true;
}
}
private Runnable checkUnconfirmed() {
return () -> {
Collection<CorrelationData> unconfirmed =
getRabbitTemplate().getUnconfirmed(getConfirmTimeout().toMillis());
unconfirmed.forEach(correlation -> handleConfirm(correlation, false, "Confirm timed out"));
};
}
@Nullable
protected abstract RabbitTemplate getRabbitTemplate();
@Override
public synchronized void stop() {
if (this.running) {
doStop();
}
this.running = false;
if (this.confirmChecker != null) {
this.confirmChecker.cancel(false);
this.confirmChecker = null;
}
}
protected void doStart() {
@@ -526,7 +571,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
}
protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message message,
int replyCode, String replyText, String exchange, String routingKey, MessageConverter converter) {
int replyCode, String replyText, String exchange, String returnedRoutingKey, MessageConverter converter) {
Object returnedObject = converter.fromMessage(message);
AbstractIntegrationMessageBuilder<?> builder = (returnedObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) returnedObject)
@@ -537,12 +582,12 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
.setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, returnedRoutingKey);
}
Message<?> returnedMessage = builder.build();
if (this.errorMessageStrategy != null) {
returnedMessage = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException(
returnedMessage, message, replyCode, replyText, exchange, routingKey), null);
returnedMessage, message, replyCode, replyText, exchange, returnedRoutingKey), null);
}
return returnedMessage;
}

42
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -42,13 +42,19 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
private final AmqpTemplate amqpTemplate;
private volatile boolean expectReply;
private final RabbitTemplate rabbitTemplate;
private boolean expectReply;
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
this.amqpTemplate = amqpTemplate;
if (amqpTemplate instanceof RabbitTemplate) {
setConnectionFactory(((RabbitTemplate) amqpTemplate).getConnectionFactory());
this.rabbitTemplate = (RabbitTemplate) amqpTemplate;
}
else {
this.rabbitTemplate = null;
}
}
@@ -62,17 +68,23 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
return this.expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter";
}
@Override
protected RabbitTemplate getRabbitTemplate() {
return this.rabbitTemplate;
}
@Override
protected void endpointInit() {
if (getConfirmCorrelationExpression() != null) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
Assert.notNull(this.rabbitTemplate,
"RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setConfirmCallback(this);
this.rabbitTemplate.setConfirmCallback(this);
}
if (getReturnChannel() != null) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
Assert.notNull(this.rabbitTemplate,
"RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
this.rabbitTemplate.setReturnCallback(this);
}
}
@@ -99,12 +111,12 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
private void send(String exchangeName, String routingKey,
final Message<?> requestMessage, CorrelationData correlationData) {
if (this.amqpTemplate instanceof RabbitTemplate) {
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
if (this.rabbitTemplate != null) {
MessageConverter converter = this.rabbitTemplate.getMessageConverter();
org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast());
addDelayProperty(requestMessage, amqpMessage);
((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);
this.rabbitTemplate.send(exchangeName, routingKey, amqpMessage, correlationData);
}
else {
this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
@@ -118,14 +130,15 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
private AbstractIntegrationMessageBuilder<?> sendAndReceive(String exchangeName, String routingKey,
Message<?> requestMessage, CorrelationData correlationData) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
Assert.state(this.rabbitTemplate != null,
"RabbitTemplate implementation is required for publisher confirms");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
MessageConverter converter = this.rabbitTemplate.getMessageConverter();
org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast());
addDelayProperty(requestMessage, amqpMessage);
org.springframework.amqp.core.Message amqpReplyMessage =
((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage,
this.rabbitTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage,
correlationData);
if (amqpReplyMessage == null) {
@@ -142,8 +155,9 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
@Override
public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
String exchange, String routingKey) {
// safe to cast; we asserted we have a RabbitTemplate in doInit()
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
// 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);
getReturnChannel().send(returned);

8
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-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.
@@ -21,6 +21,7 @@ import org.springframework.amqp.core.AmqpReplyTimeoutException;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.integration.amqp.support.MappingUtils;
import org.springframework.integration.handler.ReplyRequiredException;
@@ -61,6 +62,11 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
return "amqp:outbound-async-gateway";
}
@Override
protected RabbitTemplate getRabbitTemplate() {
return this.template.getRabbitTemplate();
}
@Override
protected void doStart() {
super.doStart();

8
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.2.xsd
View File

@@ -544,6 +544,14 @@ property set to TRUE.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Generate a negative acknowledgment (nack) if a publisher confirm is not received within this time
in milliseconds. Default none (nacks will not be generated).
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-message-strategy" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[

17
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
View File

@@ -54,6 +54,19 @@
confirm-ack-channel="ackChannel"
error-message-strategy="ems"/>
<rabbit:template id="amqpTemplateConfirms2" connection-factory="connectionFactory"/>
<amqp:outbound-channel-adapter id="withPublisherConfirms2" channel="pcRequestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"
auto-startup="false"
amqp-template="amqpTemplateConfirms2"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel"
confirm-nack-channel="nackChannel"
confirm-timeout="2000"
error-message-strategy="ems"/>
<bean id="ems" class="org.springframework.integration.support.DefaultErrorMessageStrategy" />
<int:channel id="pcRequestChannel"/>
@@ -62,6 +75,10 @@
<int:queue/>
</int:channel>
<int:channel id="nackChannel">
<int:queue/>
</int:channel>
<amqp:outbound-channel-adapter id="withDefaultAmqpTemplateExchangeAndRoutingKey"/>
<rabbit:template id="amqpTemplateWithSuppliedExchangeAndRoutingKey" connection-factory="connectionFactory"

22
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -31,6 +31,7 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.lang.reflect.Field;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -125,9 +126,9 @@ public class AmqpOutboundChannelAdapterParserTests {
.getExpressionString()).isEqualTo("42");
assertThat(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)).isFalse();
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
final AtomicBoolean shouldBePersistent = new AtomicBoolean();
@@ -179,6 +180,19 @@ public class AmqpOutboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(endpoint, "errorMessageStrategy")).isSameAs(context.getBean("ems"));
}
@Test
public void parseWithPublisherConfirms2() {
Object eventDrivenConsumer = context.getBean("withPublisherConfirms2");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
AmqpOutboundEndpoint.class);
MessageChannel nackChannel = context.getBean("nackChannel", MessageChannel.class);
MessageChannel ackChannel = context.getBean("ackChannel", MessageChannel.class);
assertThat(TestUtils.getPropertyValue(endpoint, "confirmAckChannel")).isSameAs(ackChannel);
assertThat(TestUtils.getPropertyValue(endpoint, "confirmNackChannel")).isSameAs(nackChannel);
assertThat(TestUtils.getPropertyValue(endpoint, "confirmTimeout")).isEqualTo(Duration.ofMillis(2000));
assertThat(TestUtils.getPropertyValue(endpoint, "errorMessageStrategy")).isSameAs(context.getBean("ems"));
}
@SuppressWarnings("rawtypes")
@Test
public void amqpOutboundChannelAdapterWithinChain() {
@@ -189,9 +203,9 @@ public class AmqpOutboundChannelAdapterParserTests {
AmqpOutboundEndpoint endpoint = (AmqpOutboundEndpoint) chainHandlers.get(0);
assertThat(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode")).isNull();
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(invocation -> {

16
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -109,9 +109,9 @@ public class AmqpOutboundGatewayParserTests {
assertThat(TestUtils.getPropertyValue(endpoint, "requiresReply", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)).isTrue();
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
final AtomicBoolean shouldBePersistent = new AtomicBoolean();
@@ -172,9 +172,9 @@ public class AmqpOutboundGatewayParserTests {
assertThat(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode")).isNull();
assertThat(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)).isFalse();
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(invocation -> {
@@ -221,9 +221,9 @@ public class AmqpOutboundGatewayParserTests {
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(invocation -> {
@@ -270,9 +270,9 @@ public class AmqpOutboundGatewayParserTests {
AmqpOutboundEndpoint endpoint = (AmqpOutboundEndpoint) chainHandlers.get(0);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(invocation -> {

56
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
View File

@@ -17,12 +17,22 @@
package org.springframework.integration.amqp.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.willDoNothing;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -31,15 +41,21 @@ import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.amqp.support.NackedAmqpMessageException;
import org.springframework.integration.amqp.support.ReturnedAmqpMessageException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.context.ContextConfiguration;
@@ -133,7 +149,7 @@ public class AmqpOutboundEndpointTests {
}
@Test
public void adapterWithPublisherConfirms() throws Exception {
public void adapterWithPublisherConfirms() {
Message<?> message = MessageBuilder.withPayload("hello")
.setHeader("amqp_confirmCorrelationData", "foo")
.build();
@@ -144,6 +160,42 @@ public class AmqpOutboundEndpointTests {
assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
}
@Test
public void syncConfirmTimeout() {
Message<?> message = new GenericMessage<>("foo");
RabbitTemplate template = spy(RabbitTemplate.class);
willDoNothing().given(template).send(isNull(), isNull(), any(), any());
List<CorrelationData> correlationList = new ArrayList<>();
willReturn(correlationList).given(template).getUnconfirmed(100L);
ArgumentCaptor<CorrelationData> correlationCaptor = ArgumentCaptor.forClass(CorrelationData.class);
AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(template);
PollableChannel nacks = new QueueChannel();
endpoint.setConfirmNackChannel(nacks);
endpoint.setConfirmCorrelationExpressionString("headers.id");
endpoint.setBeanFactory(mock(BeanFactory.class));
ThreadPoolTaskScheduler sched = new ThreadPoolTaskScheduler();
endpoint.setTaskScheduler(sched);
endpoint.setConfirmTimeout(100);
sched.afterPropertiesSet();
endpoint.setTaskScheduler(sched);
endpoint.afterPropertiesSet();
endpoint.start();
endpoint.handleMessage(message);
verify(template).send(isNull(), isNull(), any(), correlationCaptor.capture());
CorrelationData correlation = correlationCaptor.getValue();
correlationList.add(correlation);
assertThat(TestUtils.getPropertyValue(correlation, "message", Message.class)).isSameAs(message);
Message<?> nack = nacks.receive(10_000);
assertThat(nack).isNotNull();
assertThat(nack.getPayload()).isInstanceOf(NackedAmqpMessageException.class);
assertThat(((NackedAmqpMessageException) nack.getPayload()).getFailedMessage()).isSameAs(message);
assertThat(((NackedAmqpMessageException) nack.getPayload()).getCorrelationData())
.isSameAs(message.getHeaders().getId());
assertThat(((NackedAmqpMessageException) nack.getPayload()).getNackReason()).isEqualTo("Confirm timed out");
endpoint.stop();
sched.destroy();
}
@Test
public void adapterWithReturns() throws Exception {
this.withReturns.setErrorMessageStrategy(null);
@@ -162,7 +214,7 @@ public class AmqpOutboundEndpointTests {
}
@Test
public void adapterWithReturnsAndErrorMessageStrategy() throws Exception {
public void adapterWithReturnsAndErrorMessageStrategy() {
Message<?> message = MessageBuilder.withPayload("hello").build();
this.returnRequestChannel.send(message);
Message<?> returned = returnChannel.receive(10000);

2
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java
View File

@@ -106,7 +106,7 @@ public class AsyncAmqpGatewayTests {
try {
waitForAckBeforeReplying.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
return foo.toUpperCase();

65
src/reference/asciidoc/amqp.adoc
View File

@@ -491,6 +491,15 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
----
====
[[amqp-outbound-endpoints]]
=== Outbound Channel Adapter
The following outbound endpoints have many similar configuration options.
Starting with version 5.2, the `confirm-timeout` has been added.
Normally, when publisher confirms are enabled, the broker will quickly return an ack (or nack) which will be sent to the appropriate channel.
If a channel is closed before the confirm is received, the Spring AMQP framework will synthesize a nack.
"Missing" acks should never occur but, if you set this property, the endpoint will periodically check for them and synthesize a nack if the time elapses without a confirm being received.
[[amqp-outbound-channel-adapter]]
=== Outbound Channel Adapter
@@ -511,11 +520,12 @@ The following example shows the available properties for an AMQP outbound channe
confirm-correlation-expression="" <10>
confirm-ack-channel="" <11>
confirm-nack-channel="" <12>
return-channel="" <13>
error-message-strategy="" <14>
header-mapper="" <15>
mapped-request-headers="" <16>
lazy-connect="true" /> <17>
confirm-timeout="" <13>
return-channel="" <14>
error-message-strategy="" <15>
header-mapper="" <16>
mapped-request-headers="" <17>
lazy-connect="true" /> <18>
----
@@ -570,21 +580,24 @@ The payload is the correlation data defined by the `confirm-correlation-expressi
If the expression is `#root` or `#this`, the message is built from the original message, with the `amqp_publishConfirm` header set to `false`.
When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `NackedAmqpMessageException` payload.
Optional (the default is `nullChannel`).
<13> The channel to which returned messages are sent.
<13> When set, the adapter will synthesize a negative acknowledgment (nack) if a publisher confirm is not received within this time in milliseconds.
Pending confirms are checked every 50% of this value, so the actual time a nack is sent will be between 1x and 1.5x this value.
Default none (nacks will not be generated).
<14> The channel to which returned messages are sent.
When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter.
When there is no `ErrorMessageStrategy` configured, the message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, `amqp_returnRoutingKey`.
When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `ReturnedAmqpMessageException` payload.
Optional.
<14> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages.
<15> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages.
<15> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages.
<16> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages.
By default, only standard AMQP properties (such as `contentType`) are copied to the Spring Integration `MessageHeaders`.
Any user-defined headers is not copied to the message by the default`DefaultAmqpHeaderMapper`.
Not allowed if 'request-header-names' is provided.
Optional.
<16> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message.
<17> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message.
Not allowed if the 'header-mapper' reference is provided.
The values in this list can also be simple patterns to be matched against the header names (e.g. `"\*"` or `"thing1*, thing2"` or `"*thing1"`).
<17> When set to `false`, the endpoint attempts to connect to the broker during application context initialization.
<18> When set to `false`, the endpoint attempts to connect to the broker during application context initialization.
This allows "`fail fast`" detection of bad configuration but also causes initialization to fail if the broker is down.
When `true` (the default), the connection is established (if it does not already exist because some other component established it) when the first message is sent.
====
@@ -691,7 +704,7 @@ The following listing shows the possible properties for an AMQP Outbound Gateway
====
[source,xml]
----
<int-amqp:outbound-gateway id="inboundGateway" <1>
<int-amqp:outbound-gateway id="outboundGateway" <1>
request-channel="myRequestChannel" <2>
amqp-template="" <3>
exchange-name="" <4>
@@ -706,9 +719,10 @@ The following listing shows the possible properties for an AMQP Outbound Gateway
confirm-correlation-expression="" <13>
confirm-ack-channel="" <14>
confirm-nack-channel="" <15>
return-channel="" <16>
error-message-strategy="" <17>
lazy-connect="true" /> <18>
confirm-timeout="" <16>
return-channel="" <17>
error-message-strategy="" <18>
lazy-connect="true" /> <19>
----
@@ -771,13 +785,16 @@ The payload is the correlation data defined by `confirm-correlation-expression`
If the expression is `#root` or `#this`, the message is built from the original message, with the `amqp_publishConfirm` header set to `false`.
When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `NackedAmqpMessageException` payload.
Optional (the default is `nullChannel`).
<16> The channel to which returned messages are sent.
<16> When set, the gateway will synthesize a negative acknowledgment (nack) if a publisher confirm is not received within this time in milliseconds.
Pending confirms are checked every 50% of this value, so the actual time a nack is sent will be between 1x and 1.5x this value.
Default none (nacks will not be generated).
<17> The channel to which returned messages are sent.
When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter.
When there is no `ErrorMessageStrategy` configured, the message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, and `amqp_returnRoutingKey`.
When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `ReturnedAmqpMessageException` payload.
Optional.
<17> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages.
<18> When set to `false`, the endpoint attempts to connect to the broker during application context initialization.
<18> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages.
<19> When set to `false`, the endpoint attempts to connect to the broker during application context initialization.
This allows "`fail fast`" detection of bad configuration by logging an error message if the broker is down.
When `true` (the default), the connection is established (if it does not already exist because some other component established it) when the first message is sent.
====
@@ -901,7 +918,7 @@ The following listing shows the possible configuration options for an AMQP async
====
[source,xml]
----
<int-amqp:outbound-gateway id="inboundGateway" <1>
<int-amqp:outbound-async-gateway id="asyncOutboundGateway" <1>
request-channel="myRequestChannel" <2>
async-template="" <3>
exchange-name="" <4>
@@ -916,8 +933,9 @@ The following listing shows the possible configuration options for an AMQP async
confirm-correlation-expression="" <13>
confirm-ack-channel="" <14>
confirm-nack-channel="" <15>
return-channel="" <16>
lazy-connect="true" /> <17>
confirm-timeout="" <16>
return-channel="" <17>
lazy-connect="true" /> <18>
----
@@ -978,12 +996,15 @@ The channel to which negative (`nack`) publisher confirmations are sent.
The payload is the correlation data defined by the `confirm-correlation-expression`.
Requires the underlying `AsyncRabbitTemplate` to have its `enableConfirms` property set to `true`.
Optional (the default is `nullChannel`).
<16> The channel to which returned messages are sent.
<16> When set, the gateway will synthesize a negative acknowledgment (nack) if a publisher confirm is not received within this time in milliseconds.
Pending confirms are checked every 50% of this value, so the actual time a nack is sent will be between 1x and 1.5x this value.
Default none (nacks will not be generated).
<17> The channel to which returned messages are sent.
When provided, the underlying AMQP template is configured to return undeliverable messages to the gateway.
The message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, and `amqp_returnRoutingKey`.
Requires the underlying `AsyncRabbitTemplate` to have its `mandatory` property set to `true`.
Optional.
<17> When set to `false`, the endpoint tries to connect to the broker during application context initialization.
<18> When set to `false`, the endpoint tries to connect to the broker during application context initialization.
Doing so allows "`fail fast`" detection of bad configuration, by logging an error message if the broker is down.
When `true` (the default), the connection is established (if it does not already exist because some other component established
it) when the first message is sent.

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -16,6 +16,12 @@ See <<rate-limiter-advice>> for more information.
[[x5.2-general]]
=== General Changes
[[x5.2-amqp]]
==== AMQP Changes
The outbound endpoints can now be configured to synthesize a "nack" if no publisher confirm is received within a timeout.
See <<amqp-outbound-endpoints>> for more information.
[[x5.2-file]]
==== File Changes