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 b31a1f2a11..0fbaed3c39 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 @@ -405,7 +405,11 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler headers.put(AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE, cause); } - Message confirmMessage = this.getMessageBuilderFactory().withPayload(userCorrelationData) + AbstractIntegrationMessageBuilder builder = userCorrelationData instanceof Message + ? this.getMessageBuilderFactory().fromMessage((Message) userCorrelationData) + : this.getMessageBuilderFactory().withPayload(userCorrelationData); + + Message confirmMessage = builder .copyHeaders(headers) .build(); if (ack && this.confirmAckChannel != null) { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java index 1599d3270d..c0e0ce2930 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java @@ -147,7 +147,8 @@ public class AmqpInboundGatewayParserTests { @Test public void testInt2971HeaderMapperAndMappedHeadersExclusivity() { try { - new ClassPathXmlApplicationContext("AmqpInboundGatewayParserTests-headerMapper-fail-context.xml", this.getClass()); + new ClassPathXmlApplicationContext("AmqpInboundGatewayParserTests-headerMapper-fail-context.xml", + this.getClass()).close(); } catch (BeanDefinitionParsingException e) { assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml index f4f2c433ee..972e5a9b13 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml @@ -93,6 +93,19 @@ + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index f41f700f66..bd36379c6d 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -339,7 +339,7 @@ public class AmqpOutboundGatewayParserTests { public void testInt2971HeaderMapperAndMappedHeadersExclusivity() { try { new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-headerMapper-fail-context.xml", - this.getClass()); + this.getClass()).close();; } catch (BeanDefinitionParsingException e) { assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " + @@ -371,6 +371,19 @@ public class AmqpOutboundGatewayParserTests { assertNotNull(ack); assertEquals("foo", ack.getPayload()); assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); + + // test whole message is correlation + requestChannel = context.getBean("pcMessageCorrelationRequestChannel", MessageChannel.class); + message = MessageBuilder.withPayload("hello") + .build(); + requestChannel.send(message); + publisherCallbackChannel.handleAck(0, false); + ack = ackChannel.receive(1000); + assertNotNull(ack); + assertSame(message.getPayload(), ack.getPayload()); + assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); + + context.close(); } public static class FooAdvice extends AbstractRequestHandlerAdvice { diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index db37cb591e..7334b2ff70 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -7,24 +7,24 @@ Spring Integration provides Channel Adapters for receiving and sending messages using the Advanced Message Queuing Protocol (AMQP). The following adapters are available: -* Inbound Channel Adapter -* Outbound Channel Adapter -* Inbound Gateway -* Outbound Gateway +* <> +* <> +* <> +* <> Spring Integration also provides a point-to-point Message Channel as well as a publish/subscribe Message Channel backed by AMQP Exchanges and Queues. -In order to provide AMQP support, Spring Integration relies on Spring AMQP (http://www.springsource.org/spring-amqp[http://www.springsource.org/spring-amqp]) which "applies core Spring concepts to the development of AMQP-based messaging solutions". -Spring AMQP provides similar semantics as Spring JMS (http://static.springsource.org/spring/docs/current/spring-framework-reference/html/jms.html[http://static.springsource.org/spring/docs/current/spring-framework-reference/html/jms.html]). +In order to provide AMQP support, Spring Integration relies on (http://projects.spring.io/spring-amqp[Spring AMQP]) +which "applies core Spring concepts to the development of AMQP-based messaging solutions". +Spring AMQP provides similar semantics to (http://docs.spring.io/spring/docs/current/spring-framework-reference/html/jms.html[Spring JMS]). Whereas the provided AMQP Channel Adapters are intended for unidirectional Messaging (send or receive) only, Spring Integration also provides inbound and outbound AMQP Gateways for request/reply operations. [TIP] ===== -Please familiarize yourself with the reference documentation of the Spring AMQP project as well. +Please familiarize yourself with the +http://docs.spring.io/spring-amqp/reference/html/[reference documentation of the Spring AMQP project as well]. It provides much more in-depth information regarding Spring's integration with AMQP in general and RabbitMQ in particular. - -You can find the documentation at: http://static.springsource.org/spring-amqp/reference/html/[http://static.springsource.org/spring-amqp/reference/html/] ===== [[amqp-inbound-channel-adapter]] === Inbound Channel Adapter @@ -385,11 +385,13 @@ Requires a dedicated `RabbitTemplate` and a `CachingConnectionFactory` with the `true`. When a publisher confirm 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_. +Examples: "`headers['myCorrelationData']`", "`payload`". Starting with _version 4.1_ the `amqp_publishConfirmNackCause` message header has been added. It contains the `cause` of a 'nack' for publisher confirms. - +Starting with _version 4.2_, if the expression resolves to a `Message` instance (such as "`#this`"), the message +emitted on the ack/nack channel is based on that message, with the additional header(s) added. +Previously, a new message was created with the correlation data as its payload, regardless of type. +_Optional_. <11> The channel to which positive (ack) publisher confirms are sent; payload is the correlation data defined by the _confirm-correlation-expression_. _Optional, default=nullChannel_. @@ -407,11 +409,11 @@ _Optional_. <14> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages. -_Optional_. By default only standard AMQP properties (e.g. `contentType`) will be copied to the Spring Integration `MessageHeaders`. Any user-defined headers will NOT be copied to the Message by the default`DefaultAmqpHeaderMapper`. Not allowed if 'request-header-names' is provided. +_Optional_. <15> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message. @@ -522,10 +524,12 @@ Requires a dedicated `RabbitTemplate` and a `CachingConnectionFactory` with the `true`. When a publisher confirm 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). +For nacks, an additional header `amqp_publishConfirmNackCause` is provided. Examples: "headers['myCorrelationData']", "payload". +If the expression resolves to a `Message` instance (such as "`#this`"), the message +emitted on the ack/nack channel is based on that message, with the additional header(s) added. +Previously, a new message was created with the correlation data as its payload, regardless of type. _Optional_. -Starting with _version 4.1_ the `amqp_publishConfirmNackCause` message header has been added. -It contains the `cause` of a 'nack' for publisher confirms. <14> Since _version 4.2_. The channel to which positive (ack) publisher confirms are sent; payload is the correlation data defined by the _confirm-correlation-expression_. _Optional, default=nullChannel_. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e2448c9d50..0ff93467b7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -201,9 +201,15 @@ See <> for more information. [[x4.2-amqp-changes]] ==== AMQP Changes +===== Publisher Confirms + The `` now supports `confirm-correlation-expression` and `confirm-(n)ack-channel` attributes with similar purpose as for ``. +For both the outbound channel adapter and gateway, if the correlation data is a `Message`, it will be the basis +of the message on the ack/nack channel, with the additional header(s) added. +Previously, any correlation data (including `Message`) was returned as the payload of the ack/nack message. + See <> for more information. [[x4.2-xpath-splitter]]