From fa97ce0e6691b2ba2b382f62880f8c2b5c62a267 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 6 Feb 2020 15:32:09 -0500 Subject: [PATCH] AMQP: Multiple Sends within an OB Channel Adapter In preparation for: https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/209 When `multiSend` is true and multiple messages are sent as the payload of a message, each message is sent within the `invoke()` method of the `RabbitTemplate`. * Fix javadoc typo * Fix DSL Spec hierarchy * Protected CTOR/fields in new and modified specs * Protected CTORs, fields for remaining AMQP Specs --- .../AmqpOutboundChannelAdapterParser.java | 3 +- .../integration/amqp/dsl/Amqp.java | 10 +-- .../dsl/AmqpAsyncOutboundGatewaySpec.java | 4 +- .../AmqpBaseInboundChannelAdapterSpec.java | 6 +- .../amqp/dsl/AmqpBaseInboundGatewaySpec.java | 6 +- .../dsl/AmqpBaseOutboundEndpointSpec.java | 19 +++++- .../AmqpInboundChannelAdapterDMLCSpec.java | 4 +- .../AmqpInboundChannelAdapterSMLCSpec.java | 4 +- .../dsl/AmqpInboundChannelAdapterSpec.java | 4 +- .../amqp/dsl/AmqpInboundGatewayDMLCSpec.java | 6 +- .../amqp/dsl/AmqpInboundGatewaySMLCSpec.java | 6 +- .../amqp/dsl/AmqpInboundGatewaySpec.java | 4 +- .../AmqpInboundPolledChannelAdapterSpec.java | 6 +- .../amqp/dsl/AmqpMessageChannelSpec.java | 4 +- .../dsl/AmqpOutboundChannelAdapterSpec.java | 47 ++++++++++++++ .../amqp/dsl/AmqpOutboundEndpointSpec.java | 20 +++--- .../amqp/dsl/AmqpOutboundGatewaySpec.java | 34 ++++++++++ .../dsl/AmqpPollableMessageChannelSpec.java | 4 +- ...mqpPublishSubscribeMessageChannelSpec.java | 4 +- .../DirectMessageListenerContainerSpec.java | 4 +- .../SimpleMessageListenerContainerSpec.java | 2 +- .../amqp/outbound/AmqpOutboundEndpoint.java | 63 +++++++++++++++++-- .../amqp/config/spring-integration-amqp.xsd | 14 +++++ .../AmqpOutboundEndpointTests-context.xml | 16 ++++- .../outbound/AmqpOutboundEndpointTests.java | 27 +++++++- src/reference/asciidoc/amqp.adoc | 7 ++- src/reference/asciidoc/whats-new.adoc | 6 ++ 27 files changed, 274 insertions(+), 60 deletions(-) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundChannelAdapterSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundGatewaySpec.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java index e5fce82f0d..8d8982176b 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -82,6 +82,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "wait-for-confirm"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "multi-send"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression", diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java index e233f75bbe..014725eff0 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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. @@ -242,8 +242,8 @@ public final class Amqp { * @param amqpTemplate the amqpTemplate. * @return the AmqpOutboundEndpointSpec. */ - public static AmqpOutboundEndpointSpec outboundAdapter(AmqpTemplate amqpTemplate) { - return new AmqpOutboundEndpointSpec(amqpTemplate, false); + public static AmqpOutboundChannelAdapterSpec outboundAdapter(AmqpTemplate amqpTemplate) { + return new AmqpOutboundChannelAdapterSpec(amqpTemplate); } /** @@ -251,8 +251,8 @@ public final class Amqp { * @param amqpTemplate the amqpTemplate. * @return the AmqpOutboundEndpointSpec. */ - public static AmqpOutboundEndpointSpec outboundGateway(AmqpTemplate amqpTemplate) { - return new AmqpOutboundEndpointSpec(amqpTemplate, true); + public static AmqpOutboundGatewaySpec outboundGateway(AmqpTemplate amqpTemplate) { + return new AmqpOutboundGatewaySpec(amqpTemplate); } /** diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpAsyncOutboundGatewaySpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpAsyncOutboundGatewaySpec.java index 37cd3d13f1..c60d7dc934 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpAsyncOutboundGatewaySpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpAsyncOutboundGatewaySpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -26,7 +26,7 @@ import org.springframework.integration.amqp.outbound.AsyncAmqpOutboundGateway; public class AmqpAsyncOutboundGatewaySpec extends AmqpBaseOutboundEndpointSpec { - AmqpAsyncOutboundGatewaySpec(AsyncRabbitTemplate template) { + protected AmqpAsyncOutboundGatewaySpec(AsyncRabbitTemplate template) { this.target = new AsyncAmqpOutboundGateway(template); this.target.setRequiresReply(true); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java index 1fe6f96050..b9f3bdd028 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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. @@ -36,9 +36,9 @@ import org.springframework.retry.support.RetryTemplate; public class AmqpBaseInboundChannelAdapterSpec> extends MessageProducerSpec { - private final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper(); + protected final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper(); // NOSONAR - AmqpBaseInboundChannelAdapterSpec(AmqpInboundChannelAdapter producer) { + protected AmqpBaseInboundChannelAdapterSpec(AmqpInboundChannelAdapter producer) { super(producer); this.target.setHeaderMapper(this.headerMapper); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java index 212d2416b7..ba198bd42d 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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,9 +40,9 @@ import org.springframework.retry.support.RetryTemplate; public class AmqpBaseInboundGatewaySpec> extends MessagingGatewaySpec { - private final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper(); + protected final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper(); // NOSONAR - AmqpBaseInboundGatewaySpec(AmqpInboundGateway gateway) { + protected AmqpBaseInboundGatewaySpec(AmqpInboundGateway gateway) { super(gateway); this.target.setHeaderMapper(this.headerMapper); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java index 6e7dd4d048..ab118c07a7 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -41,7 +41,7 @@ import org.springframework.messaging.MessageChannel; * @since 5.0 */ public abstract class -AmqpBaseOutboundEndpointSpec, E extends AbstractAmqpOutboundEndpoint> + AmqpBaseOutboundEndpointSpec, E extends AbstractAmqpOutboundEndpoint> extends MessageHandlerSpec { protected final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); // NOSONAR final @@ -299,7 +299,7 @@ AmqpBaseOutboundEndpointSpec, E ext } /** - Set the error message strategy to use for returned (or negatively confirmed) + * Set the error message strategy to use for returned (or negatively confirmed) * messages. * @param errorMessageStrategy the strategy. * @return the spec. @@ -311,4 +311,17 @@ AmqpBaseOutboundEndpointSpec, E ext return _this(); } + /** + * 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 timeout the approximate timeout. + * @return the spec. + * @since 5.3 + */ + public S confirmTimeout(long timeout) { + this.target.setConfirmTimeout(timeout); + return _this(); + } + } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java index 1d89a8152f..5b96a3dde8 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -32,7 +32,7 @@ import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; public class AmqpInboundChannelAdapterDMLCSpec extends AmqpInboundChannelAdapterSpec { - AmqpInboundChannelAdapterDMLCSpec(DirectMessageListenerContainer listenerContainer) { + protected AmqpInboundChannelAdapterDMLCSpec(DirectMessageListenerContainer listenerContainer) { super(new DirectMessageListenerContainerSpec(listenerContainer)); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java index a382425066..85465546d5 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -32,7 +32,7 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; public class AmqpInboundChannelAdapterSMLCSpec extends AmqpInboundChannelAdapterSpec { - AmqpInboundChannelAdapterSMLCSpec(SimpleMessageListenerContainer listenerContainer) { + protected AmqpInboundChannelAdapterSMLCSpec(SimpleMessageListenerContainer listenerContainer) { super(new SimpleMessageListenerContainerSpec(listenerContainer)); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java index 4f7944d551..057be05453 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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,7 +42,7 @@ public abstract class AmqpInboundChannelAdapterSpec protected final AbstractMessageListenerContainerSpec listenerContainerSpec; // NOSONAR final - AmqpInboundChannelAdapterSpec(AbstractMessageListenerContainerSpec listenerContainerSpec) { + protected AmqpInboundChannelAdapterSpec(AbstractMessageListenerContainerSpec listenerContainerSpec) { super(new AmqpInboundChannelAdapter(listenerContainerSpec.get())); this.listenerContainerSpec = listenerContainerSpec; } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java index 9af759d2e0..ec7f6f48da 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -33,11 +33,11 @@ import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; public class AmqpInboundGatewayDMLCSpec extends AmqpInboundGatewaySpec { - AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { + protected AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { super(new DirectMessageListenerContainerSpec(listenerContainer), amqpTemplate); } - AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer) { + protected AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer) { super(new DirectMessageListenerContainerSpec(listenerContainer)); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java index 4925edcdf9..50f53dad0a 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -33,11 +33,11 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; public class AmqpInboundGatewaySMLCSpec extends AmqpInboundGatewaySpec { - AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { + protected AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { super(new SimpleMessageListenerContainerSpec(listenerContainer), amqpTemplate); } - AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer) { + protected AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer) { super(new SimpleMessageListenerContainerSpec(listenerContainer)); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java index aa16d36f20..a306970541 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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,7 +42,7 @@ public abstract class AmqpInboundGatewaySpec protected final AbstractMessageListenerContainerSpec listenerContainerSpec; // NOSONAR final - AmqpInboundGatewaySpec(AbstractMessageListenerContainerSpec listenerContainerSpec) { + protected AmqpInboundGatewaySpec(AbstractMessageListenerContainerSpec listenerContainerSpec) { super(new AmqpInboundGateway(listenerContainerSpec.get())); this.listenerContainerSpec = listenerContainerSpec; } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java index 2f85b0d9f6..537a7ef7c7 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,11 +35,11 @@ import org.springframework.integration.dsl.MessageSourceSpec; public class AmqpInboundPolledChannelAdapterSpec extends MessageSourceSpec { - AmqpInboundPolledChannelAdapterSpec(ConnectionFactory connectionFactory, String queue) { + protected AmqpInboundPolledChannelAdapterSpec(ConnectionFactory connectionFactory, String queue) { this.target = new AmqpMessageSource(connectionFactory, queue); } - AmqpInboundPolledChannelAdapterSpec(ConnectionFactory connectionFactory, + protected AmqpInboundPolledChannelAdapterSpec(ConnectionFactory connectionFactory, AmqpAckCallbackFactory ackCallbackFactory, String queue) { this.target = new AmqpMessageSource(connectionFactory, ackCallbackFactory, queue); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java index a0765cbd3f..46ebf92616 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java @@ -44,9 +44,9 @@ import org.springframework.util.ErrorHandler; */ public class AmqpMessageChannelSpec> extends AmqpPollableMessageChannelSpec { - private final List adviceChain = new LinkedList<>(); + protected final List adviceChain = new LinkedList<>(); // NOSONAR - AmqpMessageChannelSpec(ConnectionFactory connectionFactory) { + protected AmqpMessageChannelSpec(ConnectionFactory connectionFactory) { super(new AmqpChannelFactoryBean(true), connectionFactory); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundChannelAdapterSpec.java new file mode 100644 index 0000000000..7107f0a389 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundChannelAdapterSpec.java @@ -0,0 +1,47 @@ +/* + * Copyright 2020 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.dsl; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.messaging.Message; + +/** + * Spec for an outbound AMQP channel adapter + * + * @author Gary Russell + * @since 5.3 + * + */ +public class AmqpOutboundChannelAdapterSpec extends AmqpOutboundEndpointSpec { + + protected AmqpOutboundChannelAdapterSpec(AmqpTemplate amqpTemplate) { + super(amqpTemplate, false); + } + + /** + * If true, and the message payload is an {@link Iterable} of {@link Message}, send the + * messages in a single invocation of the template (same channel) and optionally + * wait for the confirms or die. + * @param multiSend true to send multiple messages. + * @return the spec. + */ + public AmqpOutboundChannelAdapterSpec multiSend(boolean multiSend) { + this.target.setMultiSend(multiSend); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java index 2de73b5dbe..866f6eaadb 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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,15 +21,19 @@ import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.util.Assert; /** + * Base spec for outbound AMQP endpoints. + * + * @param the spec subclass type. + * * @author Artem Bilan * @since 5.0 */ -public class AmqpOutboundEndpointSpec - extends AmqpBaseOutboundEndpointSpec { +public abstract class AmqpOutboundEndpointSpec> + extends AmqpBaseOutboundEndpointSpec { - private final boolean expectReply; + protected final boolean expectReply; // NOSONAR - AmqpOutboundEndpointSpec(AmqpTemplate amqpTemplate, boolean expectReply) { + protected AmqpOutboundEndpointSpec(AmqpTemplate amqpTemplate, boolean expectReply) { this.expectReply = expectReply; this.target = new AmqpOutboundEndpoint(amqpTemplate); this.target.setExpectReply(expectReply); @@ -40,7 +44,7 @@ public class AmqpOutboundEndpointSpec } @Override - public AmqpOutboundEndpointSpec mappedReplyHeaders(String... headers) { + public S mappedReplyHeaders(String... headers) { Assert.isTrue(this.expectReply, "'mappedReplyHeaders' can be applied only for gateway"); return super.mappedReplyHeaders(headers); } @@ -52,9 +56,9 @@ public class AmqpOutboundEndpointSpec * @since 5.2 * @see AmqpOutboundEndpoint#setWaitForConfirm(boolean) */ - public AmqpOutboundEndpointSpec waitForConfirm(boolean waitForConfirm) { + public S waitForConfirm(boolean waitForConfirm) { this.target.setWaitForConfirm(waitForConfirm); - return this; + return _this(); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundGatewaySpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundGatewaySpec.java new file mode 100644 index 0000000000..05477fb99a --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundGatewaySpec.java @@ -0,0 +1,34 @@ +/* + * Copyright 2020 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.dsl; + +import org.springframework.amqp.core.AmqpTemplate; + +/** + * Spec for an outbound AMQP gateway. + * + * @author Gary Russell + * @since 5.3 + * + */ +public class AmqpOutboundGatewaySpec extends AmqpOutboundEndpointSpec { + + protected AmqpOutboundGatewaySpec(AmqpTemplate amqpTemplate) { + super(amqpTemplate, true); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java index de06e9d1b3..510459f516 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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. @@ -44,7 +44,7 @@ public class AmqpPollableMessageChannelSpec { - AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) { + protected AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) { super(connectionFactory); this.amqpChannelFactoryBean.setPubSub(true); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java index b30d7dddd9..c5541144d2 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,7 +28,7 @@ import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; public class DirectMessageListenerContainerSpec extends AbstractMessageListenerContainerSpec { - private final DirectMessageListenerContainer listenerContainer; + protected final DirectMessageListenerContainer listenerContainer; // NOSONAR public DirectMessageListenerContainerSpec(DirectMessageListenerContainer listenerContainer) { super(listenerContainer); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java index 36994e1b9b..00e4331a9b 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java @@ -30,7 +30,7 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; public class SimpleMessageListenerContainerSpec extends AbstractMessageListenerContainerSpec { - private final SimpleMessageListenerContainer listenerContainer; + protected final SimpleMessageListenerContainer listenerContainer; // NOSONAR public SimpleMessageListenerContainerSpec(SimpleMessageListenerContainer listenerContainer) { super(listenerContainer); 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 1aea460ce5..6c1bda02cb 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -62,6 +62,8 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint private Duration waitForConfirmTimeout = DEFAULT_CONFIRM_TIMEOUT; + private boolean multiSend; + public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) { Assert.notNull(amqpTemplate, "amqpTemplate must not be null"); this.amqpTemplate = amqpTemplate; @@ -74,6 +76,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint } } + /** + * Set to true if this endpoint is a gateway. + * @param expectReply true for a gateway. + */ public void setExpectReply(boolean expectReply) { this.expectReply = expectReply; } @@ -87,6 +93,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint * @param waitForConfirm true to block until the confirmation or timeout is received. * @since 5.2 * @see #setConfirmTimeout(long) + * @see #setMultiSend(boolean) */ public void setWaitForConfirm(boolean waitForConfirm) { this.waitForConfirm = waitForConfirm; @@ -97,6 +104,23 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint return this.expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter"; } + /** + * If true, and the message payload is an {@link Iterable} of {@link Message}, send + * the messages in a single invocation of the template (same channel) and optionally + * wait for the confirms or die or perform all sends within a transaction (existing or + * new). + * @param multiSend true to send multiple messages. + * @since 5.3 + * @see #setWaitForConfirm(boolean) + */ + public void setMultiSend(boolean multiSend) { + Assert.isTrue(this.rabbitTemplate != null + && (!this.waitForConfirm || this.rabbitTemplate.getConnectionFactory().isSimplePublisherConfirms()), + () -> "To use multiSend, " + AmqpOutboundEndpoint.this.amqpTemplate + + " must be a RabbitTemplate with a ConnectionFactory configured with simple confirms"); + this.multiSend = multiSend; + } + @Override public IntegrationPatternType getIntegrationPatternType() { return this.expectReply ? super.getIntegrationPatternType() : IntegrationPatternType.outbound_channel_adapter; @@ -140,6 +164,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint if (this.expectReply) { return sendAndReceive(exchangeName, routingKey, requestMessage, correlationData); } + if (this.multiSend && requestMessage.getPayload() instanceof Iterable) { + multiSend(requestMessage, exchangeName, routingKey); + return null; + } else { send(exchangeName, routingKey, requestMessage, correlationData); if (this.waitForConfirm && correlationData != null) { @@ -149,6 +177,23 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint } } + @SuppressWarnings("unchecked") + private void multiSend(Message requestMessage, String exchangeName, String routingKey) { + ((Iterable) requestMessage.getPayload()).forEach(payload -> { + Assert.state(payload instanceof Message, + "To use multiSend, the payload must be an Iterable>"); + }); + this.rabbitTemplate.invoke(template -> { + ((Iterable>) requestMessage.getPayload()).forEach(message -> { + doRabbitSend(exchangeName, routingKey, message, null, (RabbitTemplate) template); + }); + if (this.waitForConfirm) { + template.waitForConfirmsOrDie(this.waitForConfirmTimeout.toMillis()); + } + return null; + }); + } + private void waitForConfirm(Message requestMessage, CorrelationData correlationData) { try { Confirm confirm = correlationData.getFuture().get(this.waitForConfirmTimeout.toMillis(), @@ -175,11 +220,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint final Message requestMessage, CorrelationData correlationData) { 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); - this.rabbitTemplate.send(exchangeName, routingKey, amqpMessage, correlationData); + doRabbitSend(exchangeName, routingKey, requestMessage, correlationData, this.rabbitTemplate); } else { this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(), @@ -191,6 +232,16 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint } } + private void doRabbitSend(String exchangeName, String routingKey, final Message requestMessage, + CorrelationData correlationData, RabbitTemplate template) { + + MessageConverter converter = template.getMessageConverter(); + org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, + getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast()); + addDelayProperty(requestMessage, amqpMessage); + template.send(exchangeName, routingKey, amqpMessage, correlationData); + } + private AbstractIntegrationMessageBuilder sendAndReceive(String exchangeName, String routingKey, Message requestMessage, CorrelationData correlationData) { diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd index 43cb6768ca..7df60e4f73 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd @@ -77,6 +77,20 @@ + + + + + Set to true to send payloads of type Iterable<Message<?>> + as discrete messages within a single template invocation and optionally + wait for the confirms. + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml index d836961ff1..a40d69f065 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml @@ -63,6 +63,17 @@ + + + + + + + host="localhost" publisher-returns="true" confirm-type="CORRELATED" /> + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java index b2d26741ed..2e149521fd 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -26,6 +26,7 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.concurrent.TimeUnit; @@ -84,6 +85,9 @@ public class AmqpOutboundEndpointTests { @Autowired private MessageChannel pcMessageCorrelationRequestChannel; + @Autowired + private MessageChannel multiSendChannel; + @Autowired private RabbitTemplate amqpTemplateConfirms; @@ -160,6 +164,27 @@ public class AmqpOutboundEndpointTests { assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE); } + @Test + public void multiSend() throws Exception { + RabbitTemplate template = new RabbitTemplate(this.connectionFactory); + template.setDefaultReceiveQueue(this.queue.getName()); + while (template.receive() != null) { + // drain + } + Collection> messages = new ArrayList<>(); + messages.add(new GenericMessage<>("foo")); + messages.add(new GenericMessage<>("bar")); + Message message = MessageBuilder.withPayload(messages) + .build(); + this.multiSendChannel.send(message); + org.springframework.amqp.core.Message m = receive(template); + assertThat(m).isNotNull(); + assertThat(new String(m.getBody(), "UTF-8")).isEqualTo("foo"); + m = receive(template); + assertThat(m).isNotNull(); + assertThat(new String(m.getBody(), "UTF-8")).isEqualTo("bar"); + } + @Test public void syncConfirmTimeout() { Message message = new GenericMessage<>("foo"); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 5e913119df..679a76f4bf 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -552,7 +552,8 @@ The following example shows the available properties for an AMQP outbound channe error-message-strategy="" <16> header-mapper="" <17> mapped-request-headers="" <18> - lazy-connect="true" /> <19> + lazy-connect="true" <19> + multi-send="false"/> <20> ---- <1> The unique ID for this adapter. @@ -631,6 +632,10 @@ The values in this list can also be simple patterns to be matched against the he <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. +<20> When set to `true`, payloads of type `Iterable>` will be sent as discrete messages on the same channel within the scope of a single `RabbitTemplate` invocation. +Requires a `RabbitTemplate`. +When `wait-for-confirms` is true, `RabbitTemplate.waitForConfirmsOrDie()` is invoked after the messages have been sent. +With a transactional template, the sends will be performed in either a new transaction or one that has already been started (if present). ==== [IMPORTANT] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 8c97219554..b6c3a8f5cd 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -48,6 +48,12 @@ See <<./gateway.adoc/gateway-calling-default-methods,Invoking `default` Methods> Internal components (such as `_org.springframework.integration.errorLogger`) now have a shortened name when they are represented in the integration graph. See <<./graph.adoc#integration-graph,Integration Graph>> for more information. +[[x5.3-amqp]] +=== AMQP Changes + +The outbound channel adapter has a new property `multiSend` allowing multiple messages to be sent within the scope of one `RabbitTemplate` invocation. +See <<./amqp.adoc/amqp-outbound-channel-adapter,AMQP Outbound Channel Adapter>> for more information. + [[x5.3-http]] === HTTP Changes