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

AMQP: Multiple Sends within an OB Channel Adapter

Browse Source 
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
This commit is contained in:
Gary Russell
2020-02-06 15:32:09 -05:00
committed by GitHub
parent 509e8237e8
commit fa97ce0e66
27 changed files with 274 additions and 60 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-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",

10
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java
View File

@@ -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);
}
/**

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpAsyncOutboundGatewaySpec.java
View File

@@ -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, AsyncAmqpOutboundGateway> {
AmqpAsyncOutboundGatewaySpec(AsyncRabbitTemplate template) {
protected AmqpAsyncOutboundGatewaySpec(AsyncRabbitTemplate template) {
this.target = new AsyncAmqpOutboundGateway(template);
this.target.setRequiresReply(true);
}

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java
View File

@@ -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<S extends AmqpBaseInboundChannelAdapterSpec<S>>
extends MessageProducerSpec<S, AmqpInboundChannelAdapter> {
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);
}

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java
View File

@@ -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<S extends AmqpBaseInboundGatewaySpec<S>>
extends MessagingGatewaySpec<S, AmqpInboundGateway> {
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);
}

19
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseOutboundEndpointSpec.java
View File

@@ -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<S extends AmqpBaseOutboundEndpointSpec<S, E>, E extends AbstractAmqpOutboundEndpoint>
AmqpBaseOutboundEndpointSpec<S extends AmqpBaseOutboundEndpointSpec<S, E>, E extends AbstractAmqpOutboundEndpoint>
extends MessageHandlerSpec<S, E> {
protected final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); // NOSONAR final
@@ -299,7 +299,7 @@ AmqpBaseOutboundEndpointSpec<S extends AmqpBaseOutboundEndpointSpec<S, E>, 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<S extends AmqpBaseOutboundEndpointSpec<S, E>, 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();
}
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java
View File

@@ -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> {
AmqpInboundChannelAdapterDMLCSpec(DirectMessageListenerContainer listenerContainer) {
protected AmqpInboundChannelAdapterDMLCSpec(DirectMessageListenerContainer listenerContainer) {
super(new DirectMessageListenerContainerSpec(listenerContainer));
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java
View File

@@ -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> {
AmqpInboundChannelAdapterSMLCSpec(SimpleMessageListenerContainer listenerContainer) {
protected AmqpInboundChannelAdapterSMLCSpec(SimpleMessageListenerContainer listenerContainer) {
super(new SimpleMessageListenerContainerSpec(listenerContainer));
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java
View File

@@ -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<?, C> listenerContainerSpec; // NOSONAR final
AmqpInboundChannelAdapterSpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec) {
protected AmqpInboundChannelAdapterSpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec) {
super(new AmqpInboundChannelAdapter(listenerContainerSpec.get()));
this.listenerContainerSpec = listenerContainerSpec;
}

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java
View File

@@ -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> {
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));
}

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java
View File

@@ -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> {
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));
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java
View File

@@ -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<?, C> listenerContainerSpec; // NOSONAR final
AmqpInboundGatewaySpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec) {
protected AmqpInboundGatewaySpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec) {
super(new AmqpInboundGateway(listenerContainerSpec.get()));
this.listenerContainerSpec = listenerContainerSpec;
}

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java
View File

@@ -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, AmqpMessageSource> {
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);

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpMessageChannelSpec.java
View File

@@ -44,9 +44,9 @@ import org.springframework.util.ErrorHandler;
*/
public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends AmqpPollableMessageChannelSpec<S> {
private final List<Advice> adviceChain = new LinkedList<>();
protected final List<Advice> adviceChain = new LinkedList<>(); // NOSONAR
AmqpMessageChannelSpec(ConnectionFactory connectionFactory) {
protected AmqpMessageChannelSpec(ConnectionFactory connectionFactory) {
super(new AmqpChannelFactoryBean(true), connectionFactory);
}

47
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundChannelAdapterSpec.java Normal file
View File

@@ -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<AmqpOutboundChannelAdapterSpec> {
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;
}
}

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

@@ -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 <S> the spec subclass type.
*
* @author Artem Bilan
* @since 5.0
*/
public class AmqpOutboundEndpointSpec
extends AmqpBaseOutboundEndpointSpec<AmqpOutboundEndpointSpec, AmqpOutboundEndpoint> {
public abstract class AmqpOutboundEndpointSpec<S extends AmqpOutboundEndpointSpec<S>>
extends AmqpBaseOutboundEndpointSpec<S, AmqpOutboundEndpoint> {
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();
}
}

34
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundGatewaySpec.java Normal file
View File

@@ -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<AmqpOutboundGatewaySpec> {
protected AmqpOutboundGatewaySpec(AmqpTemplate amqpTemplate) {
super(amqpTemplate, true);
}
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java
View File

@@ -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<S extends AmqpPollableMessageChannel
protected final AmqpChannelFactoryBean amqpChannelFactoryBean; // NOSONAR final
AmqpPollableMessageChannelSpec(ConnectionFactory connectionFactory) {
protected AmqpPollableMessageChannelSpec(ConnectionFactory connectionFactory) {
this(new AmqpChannelFactoryBean(false), connectionFactory);
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPublishSubscribeMessageChannelSpec.java
View File

@@ -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.
@@ -29,7 +29,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
public class AmqpPublishSubscribeMessageChannelSpec
extends AmqpMessageChannelSpec<AmqpPublishSubscribeMessageChannelSpec> {
AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) {
protected AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) {
super(connectionFactory);
this.amqpChannelFactoryBean.setPubSub(true);
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java
View File

@@ -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<DirectMessageListenerContainerSpec, DirectMessageListenerContainer> {
private final DirectMessageListenerContainer listenerContainer;
protected final DirectMessageListenerContainer listenerContainer; // NOSONAR
public DirectMessageListenerContainerSpec(DirectMessageListenerContainer listenerContainer) {
super(listenerContainer);

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java
View File

@@ -30,7 +30,7 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
public class SimpleMessageListenerContainerSpec extends
AbstractMessageListenerContainerSpec<SimpleMessageListenerContainerSpec, SimpleMessageListenerContainer> {
private final SimpleMessageListenerContainer listenerContainer;
protected final SimpleMessageListenerContainer listenerContainer; // NOSONAR
public SimpleMessageListenerContainerSpec(SimpleMessageListenerContainer listenerContainer) {
super(listenerContainer);

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

@@ -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<Message<?>>");
});
this.rabbitTemplate.invoke(template -> {
((Iterable<Message<?>>) 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) {

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

@@ -77,6 +77,20 @@
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="multi-send">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Set to true to send payloads of type Iterable&lt;Message&lt;?&gt;&gt;
as discrete messages within a single template invocation and optionally
wait for the confirms.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

16
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
View File

@@ -63,6 +63,17 @@
<int:queue/>
</int:channel>
<rabbit:template id="multiSendTemplate" connection-factory="multiSendFactory"/>
<amqp:outbound-channel-adapter id="multiSendAdapter" channel="multiSendChannel"
exchange-name=""
routing-key="#{queue.name}"
amqp-template="multiSendTemplate"
multi-send="true"
wait-for-confirm="true"/>
<int:channel id="multiSendChannel"/>
<rabbit:template id="amqpTemplateReturns" connection-factory="connectionFactory" mandatory="true" />
<amqp:outbound-channel-adapter id="withReturns" channel="returnRequestChannel"
@@ -100,7 +111,10 @@
</bean>
<rabbit:connection-factory id="connectionFactory"
host="localhost" publisher-returns="true" publisher-confirms="true" />
host="localhost" publisher-returns="true" confirm-type="CORRELATED" />
<rabbit:connection-factory id="multiSendFactory"
host="localhost" publisher-returns="true" confirm-type="SIMPLE" />
<rabbit:admin connection-factory="connectionFactory" />

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

@@ -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<Message<?>> 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");

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

@@ -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<Message<?>>` 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]

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

@@ -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