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 6f65114b0e..d1899cc825 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
@@ -43,6 +43,11 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
String amqpTemplateRef = element.getAttribute("amqp-template");
if (!StringUtils.hasText(amqpTemplateRef)) {
amqpTemplateRef = "amqpTemplate";
+ if (StringUtils.hasText(element.getAttribute("return-channel"))
+ || StringUtils.hasText(element.getAttribute("confirm-correlation-expression"))) {
+ parserContext.getReaderContext().error("A dedicated 'amqp-template' is required when" +
+ " using publisher confirms and returns", element);
+ }
}
builder.addConstructorArgReference(amqpTemplateRef);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name", true);
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 a8c51da644..31b4b3ea70 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
@@ -322,6 +322,12 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
Object userCorrelationData = correlationData;
+ if (correlationData == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No correlation data provided for ack: " + ack + " cause:" + cause);
+ }
+ return;
+ }
if (correlationData instanceof CorrelationDataWrapper) {
userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
}
diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd
index 788eb556a5..9bff535de7 100644
--- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd
+++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd
@@ -698,7 +698,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
Flag to indicate that channels created by this component will be transactional.
- Only applies to outbound messages when 'message-driven' is 'true'.
+ Only applies to messages received from this channel when 'message-driven' is 'true'.
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
index 2abb6c8574..7652342a50 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
@@ -41,9 +41,12 @@
+
+
@@ -53,9 +56,12 @@
+
+
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
index a192309c6f..33012e93b9 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
@@ -39,8 +39,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
-import com.rabbitmq.client.AMQP.BasicProperties;
-import com.rabbitmq.client.Channel;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -86,6 +84,9 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
+import com.rabbitmq.client.AMQP.BasicProperties;
+import com.rabbitmq.client.Channel;
+
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
@@ -256,7 +257,7 @@ public class AmqpOutboundChannelAdapterParserTests {
Message> message = MessageBuilder.withPayload("hello").build();
requestChannel.send(message);
PollableChannel returnChannel = context.getBean("returnChannel", PollableChannel.class);
- RabbitTemplate template = context.getBean("amqpTemplate", RabbitTemplate.class);
+ RabbitTemplate template = context.getBean("amqpTemplateReturns", RabbitTemplate.class);
Map headers = new HashMap();
headers.put(PublisherCallbackChannel.RETURN_CORRELATION, template.getUUID());
BasicProperties properties = mock(BasicProperties.class);
@@ -360,6 +361,7 @@ public class AmqpOutboundChannelAdapterParserTests {
AmqpOutboundEndpoint handler = new AmqpOutboundEndpoint(amqpTemplate);
Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class));
new DirectFieldAccessor(handler).setPropertyValue("logger", logger);
+ doAnswer(new DoesNothing()).when(logger).error("Failed to eagerly establish the connection.", toBeThrown);
ApplicationContext context = mock(ApplicationContext.class);
handler.setApplicationContext(context);
handler.setBeanFactory(context);
diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml
index 5ca9da1b29..052219fbc5 100644
--- a/src/reference/docbook/amqp.xml
+++ b/src/reference/docbook/amqp.xml
@@ -481,18 +481,20 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
An expression defining correlation data. When provided, this configures the underlying
- amqp template to receive publisher confirms. Requires a RabbitTemplate and a
+ amqp template to receive publisher confirms. Requires a dedicated
+ RabbitTemplate and a
CachingConnectionFactory with the publisherConfirms property
set to true. When a publisher confirm
- is received, it is written to either the confirm-ack-channel, or the confirm-nack-channel,
+ is received, and correlation data is supplied,
+ it is written to either the confirm-ack-channel, or the confirm-nack-channel,
depending on the confirmation type. The payload of the confirm is the correlation data as
defined by this expression and the message will have a header 'amqp_publishConfirm' set to
true (ack) or false (nack). Examples: "headers['myCorrelationData']", "payload".
Optional.
- Starting with version 4.1 the new amqp_publishConfirmNackCause
- message header has been added. It contains a cause message of 'nack' for publisher
- confirm.
+ Starting with version 4.1 the amqp_publishConfirmNackCause
+ message header has been added. It contains the cause of a 'nack' for publisher
+ confirms.
@@ -512,8 +514,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey.
Optional.
- Using a return-channel requires a RabbitTemplate with either
- the mandatory or immediate properties set to true,
+ Using a return-channel requires a RabbitTemplate with
+ the mandatory property set to true,
and a CachingConnectionFactory
with the publisherReturns property set to true. When using multiple
outbound endpoints with returns, a separate RabbitTemplate is needed
@@ -643,8 +645,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey.
Optional.
- Using a return-channel requires a RabbitTemplate with either
- the mandatory or immediate properties set to true,
+ Using a return-channel requires a RabbitTemplate with
+ the mandatory property set to true,
and a CachingConnectionFactory
with the publisherReturns property set to true. When using multiple
outbound endpoints with returns, a separate RabbitTemplate is needed