From dc94b420eef6a8d9053a28d2bfb11a0d7e714452 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 29 Mar 2016 11:11:46 -0400 Subject: [PATCH] INT-3975: AMQP Channels: Support extract-payload JIRA: https://jira.spring.io/browse/INT-3975 Also, switch the default mapping to map all inbound headers, but don't map `x-*` on outbound (by default). This will provide apps access to important headers such as `x-death` by default, while not propagating dangerous headers outbound. Polishing - PR Comments Simple code style polishing --- .../amqp/channel/AbstractAmqpChannel.java | 97 ++++++++++++++++- .../AbstractSubscribableAmqpChannel.java | 84 +++++++++++++- .../PointToPointSubscribableAmqpChannel.java | 28 ++++- .../amqp/channel/PollableAmqpChannel.java | 52 ++++++++- .../channel/PublishSubscribeAmqpChannel.java | 27 ++++- .../amqp/config/AmqpChannelFactoryBean.java | 41 ++++++- .../amqp/config/AmqpChannelParser.java | 6 +- .../AbstractAmqpOutboundEndpoint.java | 24 ---- .../amqp/outbound/AmqpOutboundEndpoint.java | 7 +- .../outbound/AsyncAmqpOutboundGateway.java | 5 +- .../amqp/support/DefaultAmqpHeaderMapper.java | 48 +++++--- .../amqp/support/MappingUtils.java | 73 +++++++++++++ .../config/spring-integration-amqp-4.3.xsd | 39 +++++++ .../amqp/channel/ChannelTests-context.xml | 18 ++- .../amqp/channel/ChannelTests.java | 103 +++++++++++++++++- .../config/AmqpChannelParserTests-context.xml | 17 +++ .../amqp/config/AmqpChannelParserTests.java | 37 ++++++- .../support/DefaultAmqpHeaderMapperTests.java | 8 +- src/reference/asciidoc/amqp.adoc | 40 +++++-- src/reference/asciidoc/whats-new.adoc | 5 + 20 files changed, 691 insertions(+), 68 deletions(-) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java index 2217601f9b..4605d9cbae 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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. @@ -17,6 +17,11 @@ package org.springframework.integration.amqp.channel; import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.MessageDeliveryMode; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.amqp.support.MappingUtils; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -29,11 +34,48 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { private final AmqpTemplate amqpTemplate; + private final RabbitTemplate rabbitTemplate; + + private final AmqpHeaderMapper outboundHeaderMapper; + + private final AmqpHeaderMapper inboundHeaderMapper; + + private volatile boolean extractPayload; + private volatile boolean loggingEnabled = true; + private MessageDeliveryMode defaultDeliveryMode; + + /** + * Construct an instance with the supplied template and default header mappers + * used if the template is a {@link RabbitTemplate} and the message is mapped. + * @param amqpTemplate the template. + * @see #setExtractPayload(boolean) + */ AbstractAmqpChannel(AmqpTemplate amqpTemplate) { + this(amqpTemplate, DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper()); + } + + /** + * Construct an instance with the supplied template and header mappers, used + * when the message is mapped. + * @param amqpTemplate the template. + * @param outboundMapper the outbound mapper. + * @param inboundMapper the inbound mapper. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + AbstractAmqpChannel(AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { Assert.notNull(amqpTemplate, "amqpTemplate must not be null"); this.amqpTemplate = amqpTemplate; + if (amqpTemplate instanceof RabbitTemplate) { + this.rabbitTemplate = (RabbitTemplate) amqpTemplate; + } + else { + this.rabbitTemplate = null; + } + this.outboundHeaderMapper = outboundMapper; + this.inboundHeaderMapper = inboundMapper; } @Override @@ -46,6 +88,41 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { this.loggingEnabled = loggingEnabled; } + /** + * Set the delivery mode to use if the message has no + * {@value org.springframework.amqp.support.AmqpHeaders#DELIVERY_MODE} + * header and the message property was not set by the {@code MessagePropertiesConverter}. + * @param defaultDeliveryMode the default delivery mode. + * @since 4.3 + */ + public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) { + this.defaultDeliveryMode = defaultDeliveryMode; + } + + /** + * Set to true to extract the payload and map the headers; otherwise + * the entire message is converted and sent. Default false. + * @param extractPayload true to extract and map. + * @since 4.3 + */ + public void setExtractPayload(boolean extractPayload) { + if (extractPayload) { + Assert.isTrue(this.rabbitTemplate != null, "amqpTemplate must be a RabbitTemplate for 'extractPayload'"); + Assert.state(this.outboundHeaderMapper != null && this.inboundHeaderMapper != null, + "'extractPayload' requires both inbound and outbound header mappers"); + } + this.extractPayload = extractPayload; + } + + /** + * @return the extract payload. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + protected boolean isExtractPayload() { + return this.extractPayload; + } + /** * Subclasses may override this method to return an Exchange name. * By default, Messages will be sent to the no-name Direct Exchange. @@ -66,13 +143,27 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { return ""; } - AmqpTemplate getAmqpTemplate() { + protected AmqpHeaderMapper getInboundHeaderMapper() { + return this.inboundHeaderMapper; + } + + protected AmqpTemplate getAmqpTemplate() { return this.amqpTemplate; } + protected RabbitTemplate getRabbitTemplate() { + return this.rabbitTemplate; + } + @Override protected boolean doSend(Message message, long timeout) { - this.amqpTemplate.convertAndSend(this.getExchangeName(), this.getRoutingKey(), message); + if (this.extractPayload) { + this.amqpTemplate.send(getExchangeName(), getRoutingKey(), MappingUtils.mapMessage(message, + this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode)); + } + else { + this.amqpTemplate.convertAndSend(getExchangeName(), getRoutingKey(), message); + } return true; } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index f1db82ee6c..98b77ef9ad 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java @@ -16,6 +16,8 @@ package org.springframework.integration.amqp.channel; +import java.util.Map; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -31,9 +33,11 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; import org.springframework.integration.MessageDispatchingException; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; @@ -65,11 +69,44 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel private final ConnectionFactory connectionFactory; + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @see #setExtractPayload(boolean) + */ protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { this(channelName, container, amqpTemplate, false); } + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @param outboundMapper the outbound mapper. + * @param inboundMapper the inbound mapper. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, + AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + this(channelName, container, amqpTemplate, false, outboundMapper, inboundMapper); + } + + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @param isPubSub true for a pub/sub channel. + * @see #setExtractPayload(boolean) + */ protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate, boolean isPubSub) { @@ -83,6 +120,32 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel this.admin = new RabbitAdmin(this.connectionFactory); } + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @param isPubSub true for a pub/sub channel. + * @param outboundMapper the outbound mapper. + * @param inboundMapper the inbound mapper. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + protected AbstractSubscribableAmqpChannel(String channelName, + SimpleMessageListenerContainer container, + AmqpTemplate amqpTemplate, boolean isPubSub, + AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + super(amqpTemplate, outboundMapper, inboundMapper); + Assert.notNull(container, "container must not be null"); + Assert.hasText(channelName, "channel name must not be empty"); + this.channelName = channelName; + this.container = container; + this.isPubSub = isPubSub; + this.connectionFactory = container.getConnectionFactory(); + this.admin = new RabbitAdmin(this.connectionFactory); + } + /** * Specify the maximum number of subscribers supported by the * channel's dispatcher (if it is an {@link AbstractDispatcher}). @@ -131,7 +194,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel : new SimpleMessageConverter(); MessageListener listener = new DispatchingMessageListener(converter, this.dispatcher, this, this.isPubSub, - this.getMessageBuilderFactory()); + getMessageBuilderFactory(), getInboundHeaderMapper()); this.container.setMessageListener(listener); if (!this.container.isActive()) { this.container.afterPropertiesSet(); @@ -157,9 +220,11 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel private final MessageBuilderFactory messageBuilderFactory; + private final AmqpHeaderMapper inboundHeaderMapper; + private DispatchingMessageListener(MessageConverter converter, MessageDispatcher dispatcher, AbstractSubscribableAmqpChannel channel, boolean isPubSub, - MessageBuilderFactory messageBuilderFactory) { + MessageBuilderFactory messageBuilderFactory, AmqpHeaderMapper inboundHeaderMapper) { Assert.notNull(converter, "MessageConverter must not be null"); Assert.notNull(dispatcher, "MessageDispatcher must not be null"); this.converter = converter; @@ -167,6 +232,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel this.channel = channel; this.isPubSub = isPubSub; this.messageBuilderFactory = messageBuilderFactory; + this.inboundHeaderMapper = inboundHeaderMapper; } @@ -177,7 +243,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel Object converted = this.converter.fromMessage(message); if (converted != null) { messageToSend = (converted instanceof Message) ? (Message) converted - : this.messageBuilderFactory.withPayload(converted).build(); + : buildMessage(message, converted); this.dispatcher.dispatch(messageToSend); } else if (this.logger.isWarnEnabled()) { @@ -203,6 +269,18 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel "while attempting to convert and dispatch Message.", e); } } + + protected Message buildMessage(org.springframework.amqp.core.Message message, Object converted) { + AbstractIntegrationMessageBuilder messageBuilder = + this.messageBuilderFactory.withPayload(converted); + if (this.channel.isExtractPayload()) { + Map headers = + this.inboundHeaderMapper.toHeadersFromRequest(message.getMessageProperties()); + messageBuilder.copyHeaders(headers); + } + return messageBuilder.build(); + } + } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java index a16d5794f3..100c0f2d73 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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. @@ -20,6 +20,7 @@ import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; @@ -34,12 +35,37 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq private volatile String queueName; + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @see #setExtractPayload(boolean) + */ public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { super(channelName, container, amqpTemplate); } + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @param outboundMapper the outbound mapper. + * @param inboundMapper the inbound mapper. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, + AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + super(channelName, container, amqpTemplate, outboundMapper, inboundMapper); + } + + /** * Provide a Queue name to be used. If this is not provided, * the Queue's name will be the same as the channel name. diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 95e4d53ff0..6b69fa98c1 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -19,12 +19,14 @@ package org.springframework.integration.amqp.channel; import java.util.ArrayDeque; import java.util.Deque; import java.util.List; +import java.util.Map; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.channel.ExecutorChannelInterceptorAware; import org.springframework.integration.support.management.PollableChannelManagement; import org.springframework.messaging.Message; @@ -54,12 +56,35 @@ public class PollableAmqpChannel extends AbstractAmqpChannel private volatile int executorInterceptorsSize; + /** + * Construct an instance with the supplied name, template and default header mappers + * used if the template is a {@link RabbitTemplate} and the message is mapped. + * @param channelName the channel name. + * @param amqpTemplate the template. + * @see #setExtractPayload(boolean) + */ public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) { super(amqpTemplate); Assert.hasText(channelName, "channel name must not be empty"); this.channelName = channelName; } + /** + * Construct an instance with the supplied name, template and header mappers. + * @param channelName the channel name. + * @param amqpTemplate the template. + * @param outboundMapper the outbound mapper. + * @param inboundMapper the inbound mapper. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, + AmqpHeaderMapper inboundMapper) { + super(amqpTemplate, inboundMapper, outboundMapper); + Assert.hasText(channelName, "channel name must not be empty"); + this.channelName = channelName; + } + /** * Provide an explicitly configured queue name. If this is not provided, then a Queue will be created @@ -141,7 +166,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel return null; } } - Object object = getAmqpTemplate().receiveAndConvert(this.queueName); + Object object = doReceive(); if (object == null) { if (isLoggingEnabled() && logger.isTraceEnabled()) { logger.trace("postReceive on channel '" + this + "', message is null"); @@ -179,6 +204,29 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } } + + protected Object doReceive() { + if (!isExtractPayload()) { + return getAmqpTemplate().receiveAndConvert(this.queueName); + } + else { + RabbitTemplate rabbitTemplate = getRabbitTemplate(); + org.springframework.amqp.core.Message message = rabbitTemplate.receive(this.queueName); + if (message != null) { + Object payload = rabbitTemplate.getMessageConverter().fromMessage(message); + Map headers = getInboundHeaderMapper() + .toHeadersFromRequest(message.getMessageProperties()); + return getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headers) + .build(); + } + else { + return null; + } + } + } + @Override public Message receive(long timeout) { if (isLoggingEnabled() && logger.isInfoEnabled()) { diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java index bc9512c389..31425d07ee 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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,6 +28,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.BroadcastingDispatcher; @@ -46,11 +47,35 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel private volatile boolean initialized; + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template. + * @see #setExtractPayload(boolean) + */ public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { super(channelName, container, amqpTemplate, true); } + /** + * Construct an instance with the supplied name, container and template; default header + * mappers will be used if the message is mapped. + * @param channelName the channel name. + * @param container the container. + * @param amqpTemplate the template + * @param outboundMapper the outbound mapper. + * @param inboundMapper the inbound mapper. + * @see #setExtractPayload(boolean) + * @since 4.3 + */ + public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container, + AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + super(channelName, container, amqpTemplate, true, outboundMapper, inboundMapper); + } + /** * Configure the FanoutExchange instance. If this is not provided, then a diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java index 770e9dd2af..bf79757063 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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 org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.FanoutExchange; +import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; @@ -41,6 +42,8 @@ import org.springframework.integration.amqp.channel.AbstractAmqpChannel; import org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel; import org.springframework.integration.amqp.channel.PollableAmqpChannel; import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.TransactionAttribute; @@ -124,6 +127,13 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean getObjectType() { return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class; @@ -321,7 +347,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean requestMessage, MessageConverter converter) { - MessageProperties amqpMessageProperties = new MessageProperties(); - org.springframework.amqp.core.Message amqpMessage; - if (converter instanceof ContentTypeDelegatingMessageConverter) { - getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); - amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); - } - else { // See INT-3002 - map headers last if we're not using a CTDMC - amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); - getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); - } - checkDeliveryMode(requestMessage, amqpMessageProperties); - return amqpMessage; - } - - private void checkDeliveryMode(Message requestMessage, MessageProperties messageProperties) { - if (this.defaultDeliveryMode != null && - requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) { - messageProperties.setDeliveryMode(this.defaultDeliveryMode); - } - } - protected Message buildReplyMessage(MessageConverter converter, org.springframework.amqp.core.Message amqpReplyMessage) { Object replyObject = converter.fromMessage(amqpReplyMessage); 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 08d3e98822..51b36c3bc0 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 @@ -25,6 +25,7 @@ import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.Lifecycle; import org.springframework.expression.Expression; +import org.springframework.integration.amqp.support.MappingUtils; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -127,7 +128,8 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint final Message requestMessage, CorrelationData correlationData) { if (this.amqpTemplate instanceof RabbitTemplate) { MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); - org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter); + org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, + getHeaderMapper(), getDefaultDeliveryMode()); ((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData); } else { @@ -149,7 +151,8 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate, "RabbitTemplate implementation is required for publisher confirms"); MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); - org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter); + org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, + getHeaderMapper(), getDefaultDeliveryMode()); org.springframework.amqp.core.Message amqpReplyMessage = ((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage, correlationData); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index bd4b098348..b3573911ca 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -22,6 +22,7 @@ import org.springframework.amqp.rabbit.AsyncRabbitTemplate; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.integration.amqp.support.MappingUtils; import org.springframework.integration.handler.ReplyRequiredException; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -60,7 +61,9 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { @Override protected Object handleRequestMessage(Message requestMessage) { RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage), - generateRoutingKey(requestMessage), mapMessage(requestMessage, this.messageConverter)); + generateRoutingKey(requestMessage), + MappingUtils.mapMessage(requestMessage, this.messageConverter, getHeaderMapper(), + getDefaultDeliveryMode())); future.addCallback(new FutureCallback(requestMessage)); CorrelationData correlationData = generateCorrelationData(requestMessage); if (correlationData != null && future.getConfirm() != null) { diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index d6d019a128..2d58f40a45 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -119,7 +119,17 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper requestMessage, + MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode) { + MessageProperties amqpMessageProperties = new MessageProperties(); + org.springframework.amqp.core.Message amqpMessage; + if (converter instanceof ContentTypeDelegatingMessageConverter) { + headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); + amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); + } + else { // See INT-3002 - map headers last if we're not using a CTDMC + amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); + headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); + } + checkDeliveryMode(requestMessage, amqpMessageProperties, defaultDeliveryMode); + return amqpMessage; + } + + /** + * Check the delivery mode and update with the default if not already present. + * @param requestMessage the request message. + * @param messageProperties the mapped message properties. + * @param defaultDeliveryMode the default delivery mode. + */ + public static void checkDeliveryMode(Message requestMessage, MessageProperties messageProperties, + MessageDeliveryMode defaultDeliveryMode) { + if (defaultDeliveryMode != null && + requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) { + messageProperties.setDeliveryMode(defaultDeliveryMode); + } + } + +} diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd index 980a446bf0..58549adb1a 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd @@ -344,6 +344,45 @@ + + + + The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'. Overridden if the 'header-mapper' + sets the delivery mode. The 'DefaultHeaderMapper' sets the value if the + Spring Integration message header 'amqp_deliveryMode' is present. If this attribute is not supplied and + the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter' + used by the 'RabbitTemplate'. If that is not customized at all, the default is 'PERSISTENT'. + + + + + + + + + + Set to 'true' to extract the message payload and map the o.s.messaging.Message to an o.s.amqp.core.Message in + a similar manner to a pair of channel adapters. When 'false' the entire message is converted requiring either + Java serializable contents or a custom message converter. Also see inbound and outbound mapped headers. + + + + + + + The header mapper to use when sending and 'extract-payload' is true. The default mapper is + DefaultAmqpHeaderMapper.outboundMapper() which maps all headers except 'x-*'. + + + + + + + The header mapper to use when receiving and 'extract-payload' is true. The default mapper is + DefaultAmqpHeaderMapper.inboundMapper() which maps all headers. + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml index bea6aa7087..bea75823e0 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml @@ -9,8 +9,24 @@ http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> - + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java index ec295b3065..d00ce94db2 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -16,6 +16,7 @@ package org.springframework.integration.amqp.channel; +import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -43,11 +44,13 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.amqp.config.AmqpChannelFactoryBean; import org.springframework.integration.amqp.rule.BrokerRunning; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.support.LogAdjustingTestSupport; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.DirtiesContext.ClassMode; @@ -66,11 +69,24 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class ChannelTests extends LogAdjustingTestSupport { @ClassRule - public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunning(); + public static final BrokerRunning brokerIsRunning = + BrokerRunning.isRunningWithEmptyQueues("pollableWithEP", "withEP"); @Autowired private PublishSubscribeAmqpChannel channel; + @Autowired + private PollableAmqpChannel pollableWithEP; + + @Autowired + private PointToPointSubscribableAmqpChannel withEP; + + @Autowired + private PublishSubscribeAmqpChannel pubSubWithEP; + + @Autowired + private PollableChannel out; + @Autowired private CachingConnectionFactory factory; @@ -81,9 +97,11 @@ public class ChannelTests extends LogAdjustingTestSupport { @After public void tearDown() { new RabbitAdmin(this.factory).deleteExchange("si.fanout.foo"); + brokerIsRunning.removeTestQueues(); } @Test + @DirtiesContext public void pubSubLostConnectionTest() throws Exception { final CyclicBarrier latch = new CyclicBarrier(2); channel.subscribe(new MessageHandler() { @@ -107,6 +125,7 @@ public class ChannelTests extends LogAdjustingTestSupport { this.channel.send(new GenericMessage("bar")); latch.await(10, TimeUnit.SECONDS); this.channel.destroy(); + this.pubSubWithEP.destroy(); assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size()); } @@ -179,4 +198,84 @@ public class ChannelTests extends LogAdjustingTestSupport { rabbitAdmin.deleteExchange("si.fanout.testChannel"); } + @Test + public void extractPayloadTests() throws Exception { + Foo foo = new Foo("bar"); + Message message = MessageBuilder.withPayload(foo).setHeader("baz", "qux").build(); + this.pollableWithEP.send(message); + Message received = this.pollableWithEP.receive(); + int n = 0; + while (received == null && n++ < 100) { + Thread.sleep(100); + received = this.pollableWithEP.receive(); + } + assertNotNull(received); + assertThat((Foo) received.getPayload(), equalTo(foo)); + assertThat((String) received.getHeaders().get("baz"), equalTo("qux")); + + this.withEP.send(message); + received = this.out.receive(10000); + assertNotNull(received); + assertThat((Foo) received.getPayload(), equalTo(foo)); + assertThat((String) received.getHeaders().get("baz"), equalTo("qux")); + + this.pubSubWithEP.send(message); + received = this.out.receive(10000); + assertNotNull(received); + assertThat((Foo) received.getPayload(), equalTo(foo)); + assertThat((String) received.getHeaders().get("baz"), equalTo("qux")); + } + + public static class Foo { + + private String bar; + + public Foo() { + } + + public Foo(String bar) { + this.bar = bar; + } + + public String getBar() { + return this.bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + ((this.bar == null) ? 0 : this.bar.hashCode()); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + Foo other = (Foo) obj; + if (this.bar == null) { + if (other.bar != null) { + return false; + } + } + else if (!this.bar.equals(other.bar)) { + return false; + } + return true; + } + + } + } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml index 0bfed1e2f0..434a326f84 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml @@ -22,4 +22,21 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java index 2d3f808c55..a0759b5586 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -16,9 +16,12 @@ package org.springframework.integration.amqp.config; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.List; @@ -26,8 +29,13 @@ import java.util.List; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.amqp.channel.AbstractAmqpChannel; +import org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel; +import org.springframework.integration.amqp.channel.PollableAmqpChannel; +import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; @@ -49,6 +57,15 @@ public class AmqpChannelParserTests { @Autowired private ApplicationContext context; + @Autowired + private PollableAmqpChannel pollableWithEP; + + @Autowired + private PointToPointSubscribableAmqpChannel withEP; + + @Autowired + private PublishSubscribeAmqpChannel pubSubWithEP; + @Test public void interceptor() { MessageChannel channel = context.getBean("channelWithInterceptor", MessageChannel.class); @@ -73,8 +90,26 @@ public class AmqpChannelParserTests { assertFalse(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class)); assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class)); assertTrue(TestUtils.getPropertyValue(channel, "amqpTemplate.transactional", Boolean.class)); + assertFalse(TestUtils.getPropertyValue(channel, "extractPayload", Boolean.class)); } + @Test + public void testMapping() { + checkExtract(this.pollableWithEP); + checkExtract(this.withEP); + checkExtract(this.pubSubWithEP); + assertEquals(MessageDeliveryMode.NON_PERSISTENT, + TestUtils.getPropertyValue(this.withEP, "defaultDeliveryMode")); + assertNull(TestUtils.getPropertyValue(this.pollableWithEP, "defaultDeliveryMode")); + } + + private void checkExtract(AbstractAmqpChannel channel) { + assertThat(TestUtils.getPropertyValue(channel, "outboundHeaderMapper").toString(), + containsString("Mock for AmqpHeaderMapper")); + assertThat(TestUtils.getPropertyValue(channel, "inboundHeaderMapper").toString(), + containsString("Mock for AmqpHeaderMapper")); + assertTrue(TestUtils.getPropertyValue(channel, "extractPayload", Boolean.class)); + } private static class TestInterceptor extends ChannelInterceptorAdapter { } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java index 2b5cf3c4b6..2b21e28831 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java @@ -264,24 +264,28 @@ public class DefaultAmqpHeaderMapperTests { assertNull(headers.get(AmqpHeaders.DELIVERY_MODE)); assertEquals(MessageDeliveryMode.NON_PERSISTENT, headers.get(AmqpHeaders.RECEIVED_DELIVERY_MODE)); assertEquals("bar", headers.get("foo")); - assertNull(headers.get("x-foo")); + assertEquals("bar", headers.get("x-foo")); + amqpProperties = new MessageProperties(); headers.put(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.NON_PERSISTENT); mapper.fromHeadersToReply(new MessageHeaders(headers), amqpProperties); assertEquals(MessageDeliveryMode.NON_PERSISTENT, amqpProperties.getDeliveryMode()); assertEquals("bar", amqpProperties.getHeaders().get("foo")); - + assertNull(amqpProperties.getHeaders().get("x-foo")); mapper = DefaultAmqpHeaderMapper.outboundMapper(); mapper.fromHeadersToRequest(new MessageHeaders(headers), amqpProperties); assertEquals(MessageDeliveryMode.NON_PERSISTENT, amqpProperties.getDeliveryMode()); assertEquals("bar", amqpProperties.getHeaders().get("foo")); + assertNull(amqpProperties.getHeaders().get("x-foo")); amqpProperties.setReceivedDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + amqpProperties.setHeader("x-death", "foo"); headers = mapper.toHeadersFromReply(amqpProperties); assertEquals(MessageDeliveryMode.NON_PERSISTENT, headers.get(AmqpHeaders.RECEIVED_DELIVERY_MODE)); assertNull(headers.get(AmqpHeaders.DELIVERY_MODE)); assertEquals("bar", headers.get("foo")); + assertEquals("foo", headers.get("x-death")); } } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index c78ae7c27f..b25cfe261b 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1142,7 +1142,8 @@ This applies to both the outbound channel adapter and gateway. There are two Message Channel implementations available. One is point-to-point, and the other is publish/subscribe. -Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate and SimpleMessageListenerContainer as you have seen on the Channel Adapters and Gateways. +Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate and +SimpleMessageListenerContainer as you have seen on the Channel Adapters and Gateways. However, the examples we'll show here are going to have minimal configuration. Explore the XML schema to view the available attributes. @@ -1154,7 +1155,9 @@ A point-to-point channel would look like this: Under the covers a Queue named "si.p2pChannel" would be declared, and this channel will send to that Queue (technically by sending to the no-name Direct Exchange with a routing key that matches this Queue's name). This channel will also register a consumer on that Queue. -If for some reason, you want the Queue to be "pollable" instead of message-driven, then simply provide the "message-driven" flag with a value of false: +If you want the channel to be "pollable" instead of message-driven, then simply provide the "message-driven" flag with +a value of `false`: + [source,xml] ---- @@ -1166,12 +1169,35 @@ A publish/subscribe channel would look like this: ---- -Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send to that Fanout Exchange. -This channel will also declare a server-named exclusive, auto-delete, non-durable Queue and bind that to the Fanout Exchange while registering a consumer on that Queue to receive Messages. +Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send to that +Fanout Exchange. +This channel will also declare a server-named exclusive, auto-delete, non-durable Queue and bind that to the Fanout +Exchange while registering a consumer on that Queue to receive Messages. There is no "pollable" option for a publish-subscribe-channel; it must be message-driven. -Starting with _version 4.1_ AMQP Backed Message Channels, alongside with `channel-transacted`, support `template-channel-transacted` to separate `transactional` configuration for the `AbstractMessageListenerContainer` and for the `RabbitTemplate`. -Note, previously, the `channel-transacted` was `true` by default, now it changed to `false` as standard default value for the `AbstractMessageListenerContainer`. +Starting with _version 4.1_ AMQP Backed Message Channels, alongside with `channel-transacted`, support +`template-channel-transacted` to separate `transactional` configuration for the `AbstractMessageListenerContainer` and +for the `RabbitTemplate`. +Note, previously, the `channel-transacted` was `true` by default, now it changed to `false` as standard default value +for the `AbstractMessageListenerContainer`. + +Prior to _version 4.3_, AMQP-backed channels only supported messages with `Serializable` payloads and headers. +The entire message was converted (serialized) and sent to RabbitMQ. +Now, you can set the `extract-payload` attribute (or `setExtractPayload()` when using Java configuration) to true. +When this flag is `true`, the message payload is converted and the headers mapped, in a similar manner to when using +channel adapters. +This allows AMQP-backed channels to be used with non-serializable payloads (perhaps with another message converter +such as the `Jackson2JsonMessageConverter`). +The default mapped headers are discussed in <>. +You can modify the mapping by providing custom mappers using the `outbound-header-mapper` and `inbound-header-mapper` +attributes. +You can now also specify a `default-delivery-mode`, used to set the delivery mode when there is no +`amqp_deliveryMode` header. +By default, Spring AMQP `MessageProperties` uses `PERSISTENT` delivery mode. + +IMPORTANT: Just as with other persistence-backed channels, AMQP-backed channels are intended to provide message +persistence to avoid message loss. +They are not intended to distribute work to other peer applications; for that purpose, use channel adapters instead. ==== Configuring with Java Configuration @@ -1258,7 +1284,7 @@ properties to support that. Any user-defined headers within the AMQP http://docs.spring.io/spring-amqp/api/org/springframework/amqp/core/MessageProperties.html[MessageProperties] WILL be copied to or from an AMQP Message, unless explicitly negated by the _requestHeaderNames_ and/or _replyHeaderNames_ properties of the `DefaultAmqpHeaderMapper`. -For an inbound mapper, all `x-*` headers are not mapped by default. +For an outbound mapper, no `x-*` headers are mapped by default; see the caution below for the reason why. To override the default, and revert to the pre-4.3 behavior, use `STANDARD_REQUEST_HEADERS` and `STANDARD_REPLY_HEADERS` in the properties. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 413a882767..5dd55e85c8 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -109,6 +109,11 @@ Spring AMQP 1.6 adds support for https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/[Delayed Message Exchanges]. Header mapping now supports the headers (`amqp_delay` and `amqp_receivedDelay`) used by this feature. +===== AMQP-Backed Channels + +AMQP-backed channels now support message mapping. +See <> for more information. + ==== Redis Changes ===== List Push/Pop Direction