INT-3499 AMQP Confirms/Returns Polishing
JIRA: https://jira.spring.io/browse/INT-3499 - Suppress expected exception from log in test case - Suppress (log) 'null' payload error when no correlation data - Enforce a specific amqp template reference in the parser when using confirms/returns - Change tests to use a dedicated template for confirms/returns
This commit is contained in:
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -698,7 +698,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
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'.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
@@ -41,9 +41,12 @@
|
||||
<amqp:outbound-channel-adapter exchange-name="outboundchanneladapter.test.1"/>
|
||||
</int:chain>
|
||||
|
||||
<rabbit:template id="amqpTemplateConfirms" connection-factory="connectionFactory"/>
|
||||
|
||||
<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannel"
|
||||
exchange-name="outboundchanneladapter.test.1"
|
||||
mapped-request-headers="foo*"
|
||||
amqp-template="amqpTemplateConfirms"
|
||||
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
|
||||
confirm-ack-channel="ackChannel"/>
|
||||
|
||||
@@ -53,9 +56,12 @@
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
|
||||
<rabbit:template id="amqpTemplateReturns" connection-factory="connectionFactory" mandatory="true" />
|
||||
|
||||
<amqp:outbound-channel-adapter id="withReturns" channel="returnRequestChannel"
|
||||
exchange-name="outboundchanneladapter.test.1"
|
||||
mapped-request-headers="foo*"
|
||||
amqp-template="amqpTemplateReturns"
|
||||
return-channel="returnChannel"/>
|
||||
|
||||
<int:channel id="returnRequestChannel"/>
|
||||
|
||||
@@ -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<String, Object> headers = new HashMap<String, Object>();
|
||||
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);
|
||||
|
||||
@@ -481,18 +481,20 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
|
||||
</callout>
|
||||
<callout arearefs="amqp-outbound-channel-adapter-xml-8-co" id="amqp-outbound-channel-adapter-xml-8">
|
||||
<para>An expression defining correlation data. When provided, this configures the underlying
|
||||
amqp template to receive publisher confirms. Requires a <classname>RabbitTemplate</classname> and a
|
||||
amqp template to receive publisher confirms. Requires a dedicated
|
||||
<classname>RabbitTemplate</classname> and a
|
||||
<classname>CachingConnectionFactory</classname> with the <code>publisherConfirms</code> property
|
||||
set to <code>true</code>. 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".
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
<para>
|
||||
Starting with <emphasis>version 4.1</emphasis> the new <code>amqp_publishConfirmNackCause</code>
|
||||
message header has been added. It contains a <code>cause</code> message of 'nack' for publisher
|
||||
confirm.
|
||||
Starting with <emphasis>version 4.1</emphasis> the <code>amqp_publishConfirmNackCause</code>
|
||||
message header has been added. It contains the <code>cause</code> of a 'nack' for publisher
|
||||
confirms.
|
||||
</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-outbound-channel-adapter-xml-9-co" id="amqp-outbound-channel-adapter-xml-9">
|
||||
@@ -512,8 +514,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
|
||||
amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey</emphasis>.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
<important>
|
||||
Using a <code>return-channel</code> requires a <classname>RabbitTemplate</classname> with either
|
||||
the <code>mandatory</code> or <code>immediate</code> properties set to <code>true</code>,
|
||||
Using a <code>return-channel</code> requires a <classname>RabbitTemplate</classname> with
|
||||
the <code>mandatory</code> property set to <code>true</code>,
|
||||
and a <classname>CachingConnectionFactory</classname>
|
||||
with the <code>publisherReturns</code> property set to <code>true</code>. When using multiple
|
||||
outbound endpoints with returns, a separate <classname>RabbitTemplate</classname> is needed
|
||||
@@ -643,8 +645,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann
|
||||
amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey</emphasis>.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
<important>
|
||||
Using a <code>return-channel</code> requires a <classname>RabbitTemplate</classname> with either
|
||||
the <code>mandatory</code> or <code>immediate</code> properties set to <code>true</code>,
|
||||
Using a <code>return-channel</code> requires a <classname>RabbitTemplate</classname> with
|
||||
the <code>mandatory</code> property set to <code>true</code>,
|
||||
and a <classname>CachingConnectionFactory</classname>
|
||||
with the <code>publisherReturns</code> property set to <code>true</code>. When using multiple
|
||||
outbound endpoints with returns, a separate <classname>RabbitTemplate</classname> is needed
|
||||
|
||||
Reference in New Issue
Block a user