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

AMQP OB Adapter - option to wait for confirms

Browse Source 
- add an option to block the caller until a confirm is received

* Resolve PR comments re exceptions, default timeout etc
This commit is contained in:
Gary Russell
2019-08-15 17:18:59 -04:00
committed by Artem Bilan
parent fab4c452d2
commit f256974dd8
10 changed files with 241 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -81,6 +81,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.setValueIfAttributeDefined(builder, element, "wait-for-confirm");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression",

12
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java
View File

@@ -45,4 +45,16 @@ public class AmqpOutboundEndpointSpec
return super.mappedReplyHeaders(headers);
}
/**
* Wait for a publisher confirm.
* @param waitForConfirm true to wait.
* @return the spec.
* @since 5.2
* @see AmqpOutboundEndpoint#setWaitForConfirm(boolean)
*/
public AmqpOutboundEndpointSpec waitForConfirm(boolean waitForConfirm) {
this.target.setWaitForConfirm(waitForConfirm);
return this;
}
}

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

@@ -402,6 +402,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
return this.headersMappedLast;
}
@Nullable
protected Duration getConfirmTimeout() {
return this.confirmTimeout;
}

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

@@ -16,12 +16,21 @@
package org.springframework.integration.amqp.outbound;
import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
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.MessageTimeoutException;
import org.springframework.integration.amqp.support.MappingUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
@@ -38,7 +47,9 @@ import org.springframework.util.Assert;
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
implements RabbitTemplate.ConfirmCallback, ReturnCallback {
implements ConfirmCallback, ReturnCallback {
private static final Duration DEFAULT_CONFIRM_TIMEOUT = Duration.ofSeconds(5);
private final AmqpTemplate amqpTemplate;
@@ -46,6 +57,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
private boolean expectReply;
private boolean waitForConfirm;
private Duration waitForConfirmTimeout = DEFAULT_CONFIRM_TIMEOUT;
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
this.amqpTemplate = amqpTemplate;
@@ -62,6 +77,19 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
this.expectReply = expectReply;
}
/**
* Set to true if you want to block the calling thread until a publisher confirm has
* been received. Requires a template configured for returns. If a confirm is not
* received within the confirm timeout or a negative acknowledgment or returned
* message is received, an exception will be thrown. Does not apply to the gateway
* since it blocks awaiting the reply.
* @param waitForConfirm true to block until the confirmation or timeout is received.
* @since 5.2
* @see #setConfirmTimeout(long)
*/
public void setWaitForConfirm(boolean waitForConfirm) {
this.waitForConfirm = waitForConfirm;
}
@Override
public String getComponentType() {
@@ -86,6 +114,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
"RabbitTemplate implementation is required for publisher confirms");
this.rabbitTemplate.setReturnCallback(this);
}
Duration confirmTimeout = getConfirmTimeout();
if (confirmTimeout != null) {
this.waitForConfirmTimeout = confirmTimeout;
}
}
@Override
@@ -101,14 +133,39 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
String exchangeName = generateExchangeName(requestMessage);
String routingKey = generateRoutingKey(requestMessage);
if (this.expectReply) {
return this.sendAndReceive(exchangeName, routingKey, requestMessage, correlationData);
return sendAndReceive(exchangeName, routingKey, requestMessage, correlationData);
}
else {
this.send(exchangeName, routingKey, requestMessage, correlationData);
send(exchangeName, routingKey, requestMessage, correlationData);
if (this.waitForConfirm && correlationData != null) {
waitForConfirm(requestMessage, correlationData);
}
return null;
}
}
private void waitForConfirm(Message<?> requestMessage, CorrelationData correlationData) {
try {
Confirm confirm = correlationData.getFuture().get(this.waitForConfirmTimeout.toMillis(),
TimeUnit.MILLISECONDS);
if (!confirm.isAck()) {
throw new AmqpException("Negative publisher confirm received: " + confirm);
}
if (correlationData.getReturnedMessage() != null) {
throw new AmqpException("Message was returned by the broker");
}
}
catch (@SuppressWarnings("unused") InterruptedException e) {
Thread.currentThread().interrupt();
}
catch (ExecutionException e) {
throw new AmqpException("Failed to get publisher confirm", e);
}
catch (TimeoutException e) {
throw new MessageTimeoutException(requestMessage, this + ": Timed out awaiting publisher confirm", e);
}
}
private void send(String exchangeName, String routingKey,
final Message<?> requestMessage, CorrelationData correlationData) {
if (this.rabbitTemplate != null) {

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

@@ -58,6 +58,21 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="wait-for-confirm">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Set to true if you want to block the calling thread until a publisher confirm has
been received. Requires a template configured for returns. If a confirm is not
received within the confirm timeout or a negative acknowledgment or returned
message is received, an exception will be thrown.
</xsd:documentation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

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

@@ -65,6 +65,7 @@
confirm-ack-channel="ackChannel"
confirm-nack-channel="nackChannel"
confirm-timeout="2000"
wait-for-confirm="true"
error-message-strategy="ems"/>
<bean id="ems" class="org.springframework.integration.support.DefaultErrorMessageStrategy" />

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

@@ -178,6 +178,7 @@ public class AmqpOutboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(endpoint, "confirmAckChannel")).isSameAs(ackChannel);
assertThat(TestUtils.getPropertyValue(endpoint, "confirmNackChannel")).isSameAs(nullChannel);
assertThat(TestUtils.getPropertyValue(endpoint, "errorMessageStrategy")).isSameAs(context.getBean("ems"));
assertThat(TestUtils.getPropertyValue(endpoint, "waitForConfirm", Boolean.class)).isFalse();
}
@Test
@@ -191,6 +192,7 @@ public class AmqpOutboundChannelAdapterParserTests {
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"));
assertThat(TestUtils.getPropertyValue(endpoint, "waitForConfirm", Boolean.class)).isTrue();
}
@SuppressWarnings("rawtypes")

130
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java Normal file
View File

@@ -0,0 +1,130 @@
/*
* Copyright 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.integration.amqp.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.Collections;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
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.ConnectionFactory;
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.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.amqp.dsl.Amqp;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.DisabledIf;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @since 5.2
*
*/
@SpringJUnitConfig
@RabbitAvailable(queues = "testConfirmOk")
@DirtiesContext
public class AmqpOutboundEndpointTests2 {
@Test
void testConfirmOk(@Autowired IntegrationFlow flow, @Autowired RabbitTemplate template) {
flow.getInputChannel().send(new GenericMessage<>("test", Collections.singletonMap("rk", "testConfirmOk")));
assertThat(template.receive("testConfirmOk")).isNotNull();
}
@Test
void testWithReturn(@Autowired IntegrationFlow flow) {
assertThatThrownBy(() -> flow.getInputChannel()
.send(new GenericMessage<>("test", Collections.singletonMap("rk", "junkjunk"))))
.isInstanceOf(MessageHandlingException.class)
.hasCauseInstanceOf(AmqpException.class)
.extracting(ex -> ex.getCause())
.extracting(ex -> ex.getMessage())
.isEqualTo("Message was returned by the broker");
}
@Test
@DisabledIf("#{systemEnvironment['TRAVIS'] ?: false}") // needs RabbitMQ 3.7
void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admin,
@Autowired RabbitTemplate template) {
Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1).overflow(Overflow.rejectPublish).build();
admin.declareQueue(queue);
flow.getInputChannel().send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName())));
assertThatThrownBy(() -> flow.getInputChannel()
.send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName()))))
.hasCauseInstanceOf(AmqpException.class)
.extracting(ex -> ex.getCause())
.extracting(ex -> ex.getMessage())
.matches(msg -> msg.matches("Negative publisher confirm received: .*"));
assertThat(template.receive(queue.getName())).isNotNull();
admin.deleteQueue(queue.getName());
}
@Configuration(proxyBeanMethods = false)
@EnableIntegration
public static class Config {
@Bean
public IntegrationFlow flow(RabbitTemplate template) {
return f -> f.handle(Amqp.outboundAdapter(template)
.exchangeName("")
.routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class))
.confirmCorrelationFunction(msg -> msg)
.waitForConfirm(true));
}
@Bean
public CachingConnectionFactory cf() {
CachingConnectionFactory ccf = new CachingConnectionFactory(
RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
ccf.setPublisherConfirms(true);
ccf.setPublisherReturns(true);
return ccf;
}
@Bean
public RabbitTemplate template(ConnectionFactory cf) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(cf);
rabbitTemplate.setMandatory(true);
rabbitTemplate.setReceiveTimeout(10_000);
return rabbitTemplate;
}
@Bean
public RabbitAdmin admin(ConnectionFactory cf) {
return new RabbitAdmin(cf);
}
}
}

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

@@ -547,12 +547,12 @@ The following example shows the available properties for an AMQP outbound channe
confirm-ack-channel="" <11>
confirm-nack-channel="" <12>
confirm-timeout="" <13>
return-channel="" <14>
error-message-strategy="" <15>
header-mapper="" <16>
mapped-request-headers="" <17>
lazy-connect="true" /> <18>
wait-for-confirm="" <14>
return-channel="" <15>
error-message-strategy="" <16>
header-mapper="" <17>
mapped-request-headers="" <18>
lazy-connect="true" /> <19>
----
<1> The unique ID for this adapter.
@@ -609,21 +609,26 @@ Optional (the default is `nullChannel`).
<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.
<14> When set to true, the calling thread will block, waiting for a publisher confirmation.
This requires a `RabbitTemplate` configured for confirms as well as a `confirm-correlation-expression`.
The thread will block for up to `confirm-timeout` (or 5 seconds by default).
If a timeout occurs, a `MessageTimeoutException` will be thrown.
If returns are enabled and a message is returned, or any other exception occurs while awaiting the confirm, a `MessageHandlingException` will be thrown, with an appropriate message.
<15> 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.
<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.
<16> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages.
<17> 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.
<17> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message.
<18> 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"`).
<18> When set to `false`, the endpoint attempts to connect to the broker during application context initialization.
<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 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.
====

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

@@ -89,6 +89,9 @@ See <<./amqp.adoc#amqp-outbound-endpoints,Outbound Channel Adapter>> for more in
The inbound channel adapter can now receive batched messages as a `List<?>` payload instead of receiving a discrete message for each batch fragment.
See <<./amqp.adoc#amqp-debatching,Batched Messages>> for more information.
The outbound channel adapter can now be configured to block the calling thread until a publisher confirm (acknowledgment) is received.
See <<./amqp.adoc#amqp-outbound-channel-adapter,Outbound Channel Adapter>> for more information.
[[x5.2-file]]
==== File Changes