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

INT-3965: AMQP Header Mapping Changes

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3965

Rework According to AMQP-589
This commit is contained in:
Gary Russell
2016-03-11 22:59:13 -05:00
committed by Artem Bilan
parent 518df4f74b
commit 78aaa6dac6
18 changed files with 198 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -133,7 +133,7 @@ subprojects { subproject ->
tomcatVersion = "8.0.30"
smack3Version = '3.2.1'
smackVersion = '4.1.5'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.6.0.M1'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.6.0.BUILD-SNAPSHOT'
// springCloudClusterVersion = '1.0.0.BUILD-SNAPSHOT'
springDataJpaVersion = '1.10.0.M1'
springDataMongoVersion = '1.9.0.M1'

7
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java
View File

@@ -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.
@@ -105,7 +105,10 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
BeanDefinitionBuilder mapperBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultAmqpHeaderMapper.class);
mapperBuilder.setFactoryMethod("inboundMapper");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, mapperBuilder, null);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");

7
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -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.
@@ -67,8 +67,11 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "lazy-connect");
BeanDefinitionBuilder mapperBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultAmqpHeaderMapper.class);
mapperBuilder.setFactoryMethod("outboundMapper");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext,
DefaultAmqpHeaderMapper.class, null);
mapperBuilder, null);
BeanDefinition confirmCorrelationExpression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("confirm-correlation-expression", element);

5
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
View File

@@ -82,7 +82,10 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class,
BeanDefinitionBuilder mapperBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultAmqpHeaderMapper.class);
mapperBuilder.setFactoryMethod("outboundMapper");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, mapperBuilder,
null);
return builder;

8
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -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.
@@ -18,8 +18,6 @@ package org.springframework.integration.amqp.inbound;
import java.util.Map;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
@@ -33,6 +31,8 @@ import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.util.Assert;
import com.rabbitmq.client.Channel;
/**
* Adapter that receives Messages from an AMQP Queue, converts them into
* Spring Integration Messages, and sends the results to a Message Channel.
@@ -49,7 +49,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) {

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -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.
@@ -59,7 +59,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
private volatile MessageConverter amqpMessageConverter = new SimpleMessageConverter();
private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
private Address defaultReplyTo;

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

@@ -77,7 +77,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
private volatile ExpressionEvaluatingMessageProcessor<String> exchangeNameGenerator;
private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
private volatile Expression confirmCorrelationExpression;

58
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
View File

@@ -77,6 +77,7 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
STANDARD_HEADER_NAMES.add(AmqpHeaders.MESSAGE_COUNT);
STANDARD_HEADER_NAMES.add(AmqpHeaders.MESSAGE_ID);
STANDARD_HEADER_NAMES.add(AmqpHeaders.RECEIVED_DELAY);
STANDARD_HEADER_NAMES.add(AmqpHeaders.RECEIVED_DELIVERY_MODE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.RECEIVED_EXCHANGE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.RECEIVED_ROUTING_KEY);
STANDARD_HEADER_NAMES.add(AmqpHeaders.REDELIVERED);
@@ -112,6 +113,11 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
CONSUMER_METADATA_PRESENT = consumerTagHeader.get();
}
/**
* @deprecated - use {@link #inboundMapper()} and {@link #outboundMapper()} depending
* on the context in which the mapper is to be used.
*/
@Deprecated
public DefaultAmqpHeaderMapper() {
super(AmqpHeaders.PREFIX, STANDARD_HEADER_NAMES, STANDARD_HEADER_NAMES);
}
@@ -147,9 +153,9 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
if (correlationId != null && correlationId.length > 0) {
headers.put(AmqpHeaders.CORRELATION_ID, correlationId);
}
MessageDeliveryMode deliveryMode = amqpMessageProperties.getDeliveryMode();
if (deliveryMode != null) {
headers.put(AmqpHeaders.DELIVERY_MODE, deliveryMode);
MessageDeliveryMode receivedDeliveryMode = amqpMessageProperties.getReceivedDeliveryMode();
if (receivedDeliveryMode != null) {
headers.put(AmqpHeaders.RECEIVED_DELIVERY_MODE, receivedDeliveryMode);
}
long deliveryTag = amqpMessageProperties.getDeliveryTag();
if (deliveryTag > 0) {
@@ -405,4 +411,50 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
}
}
public static DefaultAmqpHeaderMapper inboundMapper() {
DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
mapper.setRequestHeaderNames(inboundRequestHeaders());
mapper.setReplyHeaderNames(inboundReplyHeaders());
return mapper;
}
public static DefaultAmqpHeaderMapper outboundMapper() {
DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
mapper.setRequestHeaderNames(outboundRequestHeaders());
mapper.setReplyHeaderNames(outboundReplyHeaders());
return mapper;
}
/**
* @return the default request headers for an inbound mapper.
*/
public static String[] inboundRequestHeaders() {
return safeInboundHeaders();
}
/**
* @return the default reply headers for an inbound mapper.
*/
public static String[] inboundReplyHeaders() {
return new String[] { "*" };
}
/**
* @return the default request headers for an outbound mapper.
*/
public static String[] outboundRequestHeaders() {
return new String[] { "*" };
}
/**
* @return the default reply headers for an outbound mapper.
*/
public static String[] outboundReplyHeaders() {
return safeInboundHeaders();
}
private static String[] safeInboundHeaders() {
return new String[] { "!x-*", "*" };
}
}

6
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
View File

@@ -102,6 +102,12 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
public boolean isOpen() {
return false;
}
@Override
public int getLocalPort() {
return 0;
}
}
private static class StubChannel implements Channel {

24
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests-context.xml
View File

@@ -23,22 +23,26 @@
<amqp:inbound-channel-adapter id="autoStartFalse" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123" acknowledge-mode="${ackMode}" missing-queues-fatal="false" />
<amqp:inbound-channel-adapter id="withHeaderMapperStandardAndCustomHeaders" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
<amqp:inbound-channel-adapter id="withHeaderMapperStandardAndCustomHeaders"
channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers="foo*, STANDARD_REQUEST_HEADERS"/>
<amqp:inbound-channel-adapter id="withHeaderMapperOnlyCustomHeaders" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
<amqp:inbound-channel-adapter id="withHeaderMapperOnlyCustomHeaders"
channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers="foo*"/>
<amqp:inbound-channel-adapter id="withHeaderMapperNothingToMap" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
<amqp:inbound-channel-adapter id="withHeaderMapperNothingToMap"
channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers=""/>
<amqp:inbound-channel-adapter id="withHeaderMapperDefaultMapping" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
<amqp:inbound-channel-adapter id="withHeaderMapperDefaultMapping"
channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"/>
<int:channel id="requestChannel">
<int:queue/>
</int:channel>
@@ -46,5 +50,5 @@
<bean id="rabbitConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
</bean>
</beans>

8
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java
View File

@@ -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.
@@ -187,8 +187,8 @@ public class AmqpInboundChannelAdapterParserTests {
listener.onMessage(amqpMessage, null);
QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class);
org.springframework.messaging.Message<?> siMessage = requestChannel.receive(0);
assertNull(siMessage.getHeaders().get("bar"));
assertNull(siMessage.getHeaders().get("foo"));
assertNotNull(siMessage.getHeaders().get("bar"));
assertNotNull(siMessage.getHeaders().get("foo"));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_ENCODING));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CLUSTER_ID));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.APP_ID));
@@ -199,7 +199,7 @@ public class AmqpInboundChannelAdapterParserTests {
public void testInt2971HeaderMapperAndMappedHeadersExclusivity() {
try {
new ClassPathXmlApplicationContext("AmqpInboundChannelAdapterParserTests-headerMapper-fail-context.xml",
this.getClass());
this.getClass()).close();
}
catch (BeanDefinitionParsingException e) {
assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " +

5
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -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.
@@ -186,6 +186,7 @@ public class AmqpOutboundGatewayParserTests {
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode());
amqpProperties.setReceivedDeliveryMode(properties.getDeliveryMode());
return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
}
})
@@ -207,7 +208,7 @@ public class AmqpOutboundGatewayParserTests {
assertEquals("bar", replyMessage.getHeaders().get("bar"));
assertEquals("foo", replyMessage.getHeaders().get("foo")); // copied from request Message
assertNull(replyMessage.getHeaders().get("foobar"));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.RECEIVED_DELIVERY_MODE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
context.close();

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2015 the original author or authors.
* Copyright 2013-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.
@@ -100,7 +100,7 @@ public class InboundEndpointTests {
amqpMessageProperties.setDeliveryTag(123L);
org.springframework.amqp.core.Message amqpMessage =
new SimpleMessageConverter().toMessage(jsonMessage.getPayload(), amqpMessageProperties);
new DefaultAmqpHeaderMapper().fromHeadersToRequest(jsonMessage.getHeaders(), amqpMessageProperties);
DefaultAmqpHeaderMapper.inboundMapper().fromHeadersToRequest(jsonMessage.getHeaders(), amqpMessageProperties);
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
Channel rabbitChannel = mock(Channel.class);

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java
View File

@@ -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.
@@ -77,7 +77,7 @@ public class OutboundEndpointTests {
TestRabbitTemplate amqpTemplate = spy(new TestRabbitTemplate(connectionFactory));
AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
endpoint.setExpectReply(true);
DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
mapper.setRequestHeaderNames("*");
endpoint.setHeaderMapper(mapper);
final AtomicReference<Message> amqpMessage =

51
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
View File

@@ -51,7 +51,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put(AmqpHeaders.APP_ID, "test.appId");
headerMap.put(AmqpHeaders.CLUSTER_ID, "test.clusterId");
@@ -113,7 +113,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeadersWithContentTypeAsMediaType() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
MediaType contentType = MediaType.parseMediaType("text/html");
@@ -135,7 +135,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeadersWithContentTypeAsMimeType() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
MimeType contentType = MimeType.valueOf("text/html");
@@ -151,7 +151,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void toHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
@@ -160,7 +160,7 @@ public class DefaultAmqpHeaderMapperTests {
amqpProperties.setContentType("test.contentType");
byte[] testCorrelationId = new byte[] {1, 2, 3};
amqpProperties.setCorrelationId(testCorrelationId);
amqpProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
amqpProperties.setReceivedDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
amqpProperties.setDeliveryTag(1234L);
amqpProperties.setExpiration("test.expiration");
amqpProperties.setMessageCount(42);
@@ -184,7 +184,7 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(99L, headerMap.get(AmqpHeaders.CONTENT_LENGTH));
assertEquals("test.contentType", headerMap.get(AmqpHeaders.CONTENT_TYPE));
assertEquals(testCorrelationId, headerMap.get(AmqpHeaders.CORRELATION_ID));
assertEquals(MessageDeliveryMode.NON_PERSISTENT, headerMap.get(AmqpHeaders.DELIVERY_MODE));
assertEquals(MessageDeliveryMode.NON_PERSISTENT, headerMap.get(AmqpHeaders.RECEIVED_DELIVERY_MODE));
assertEquals(1234L, headerMap.get(AmqpHeaders.DELIVERY_TAG));
assertEquals("test.expiration", headerMap.get(AmqpHeaders.EXPIRATION));
assertEquals(42, headerMap.get(AmqpHeaders.MESSAGE_COUNT));
@@ -203,7 +203,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test // INT-3586 requires Spring AMQP 1.4.2
public void testToHeadersConsumerMetadata() {
try {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setConsumerTag("consumerTag");
amqpProperties.setConsumerQueue("consumerQueue");
@@ -218,7 +218,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void messageIdNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put(MessageHeaders.ID, "msg-id");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
@@ -229,7 +229,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void messageTimestampNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put(MessageHeaders.TIMESTAMP, 1234L);
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
@@ -240,7 +240,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test // INT-2090
public void jsonTypeIdNotOverwritten() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
MessageConverter converter = new Jackson2JsonMessageConverter();
MessageProperties amqpProperties = new MessageProperties();
converter.toMessage("123", amqpProperties);
@@ -253,4 +253,35 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(String.class, result.getClass());
}
@Test
public void inboundOutbound() {
DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setReceivedDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
amqpProperties.getHeaders().put("foo", "bar");
amqpProperties.getHeaders().put("x-foo", "bar");
Map<String, Object> headers = mapper.toHeadersFromRequest(amqpProperties);
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"));
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"));
mapper = DefaultAmqpHeaderMapper.outboundMapper();
mapper.fromHeadersToRequest(new MessageHeaders(headers), amqpProperties);
assertEquals(MessageDeliveryMode.NON_PERSISTENT, amqpProperties.getDeliveryMode());
assertEquals("bar", amqpProperties.getHeaders().get("foo"));
amqpProperties.setReceivedDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
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"));
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -326,6 +326,21 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void configureHeaderMapper(Element element, BeanDefinitionBuilder rootBuilder,
ParserContext parserContext, Class<?> headerMapperClass, String replyHeaderValue) {
configureHeaderMapper(element, rootBuilder, parserContext,
BeanDefinitionBuilder.genericBeanDefinition(headerMapperClass), replyHeaderValue);
}
/**
* Utility method to configure a HeaderMapper for Inbound and Outbound channel adapters/gateway.
*
* @param element The element.
* @param rootBuilder The root builder.
* @param parserContext The parser context.
* @param headerMapperBuilder The header mapper builder.
* @param replyHeaderValue The reply header value.
*/
public static void configureHeaderMapper(Element element, BeanDefinitionBuilder rootBuilder,
ParserContext parserContext, BeanDefinitionBuilder headerMapperBuilder, String replyHeaderValue) {
String defaultMappedReplyHeadersAttributeName = "mapped-reply-headers";
if (!StringUtils.hasText(replyHeaderValue)){
replyHeaderValue = defaultMappedReplyHeadersAttributeName;
@@ -343,7 +358,6 @@ public abstract class IntegrationNamespaceUtils {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(rootBuilder, element, "header-mapper");
if (hasMappedRequestHeaders || hasMappedReplyHeaders){
BeanDefinitionBuilder headerMapperBuilder = BeanDefinitionBuilder.genericBeanDefinition(headerMapperClass);
if (hasMappedRequestHeaders) {
headerMapperBuilder.addPropertyValue("requestHeaderNames", element.getAttribute("mapped-request-headers"));
@@ -355,7 +369,6 @@ public abstract class IntegrationNamespaceUtils {
rootBuilder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition());
}
}
/**
* Parse a "transactional" element and configure a {@link TransactionInterceptor}
* with "transactionManager" and other "transactionDefinition" properties.

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

@@ -1058,23 +1058,32 @@ public IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) {
[[amqp-message-headers]]
=== AMQP Message Headers
The Spring Integration AMQP Adapters will map standard AMQP properties automatically.
The Spring Integration AMQP Adapters will map all AMQP properties and headers automatically.
(This is a change in 4.3 - previously, only standard headers were mapped).
These properties will be copied by default to and from Spring Integration `MessageHeaders` using the
http://docs.spring.io/spring-integration/api/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.html[DefaultAmqpHeaderMapper].
Of course, you can pass in your own implementation of AMQP specific header mappers, as the adapters have respective properties to support that.
Of course, you can pass in your own implementation of AMQP specific header mappers, as the adapters have respective
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 NOT
be copied to or from an AMQP Message, unless explicitly specified by the _requestHeaderNames_ and/or
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.
TIP: When mapping user-defined headers, the values can also contain simple wildcard patterns (e.g. "foo*" or "*foo") to be matched.
For example, if you need to copy all user-defined headers simply use the wildcard character `*`, but see the CAUTION below.
To override the default, and revert to the pre-4.3 behavior, use `STANDARD_REQUEST_HEADERS` and
`STANDARD_REPLY_HEADERS` in the properties.
Starting with _version 4.1_, the `AbstractHeaderMapper` (a `DefaultAmqpHeaderMapper` superclass) allows the `NON_STANDARD_HEADERS` token to be configured for the _requestHeaderNames_ and/or _replyHeaderNames_ properties (in addition to existing `STANDARD_REQUEST_HEADERS` and `STANDARD_REPLY_HEADERS`) to map all user-defined headers.
Note, it is recommended to use the combination like this `STANDARD_REPLY_HEADERS, NON_STANDARD_HEADERS` instead of generic `*`, to avoid mapping of _request_ headers to the reply.
TIP: When mapping user-defined headers, the values can also contain simple wildcard patterns (e.g. "foo*" or "*foo")
to be matched.
`*` matches all headers.
Class `org.springframework.amqp.support.AmqpHeaders` identifies the default headers that will be used by the `DefaultAmqpHeaderMapper`:
Starting with _version 4.1_, the `AbstractHeaderMapper` (a `DefaultAmqpHeaderMapper` superclass) allows the
`NON_STANDARD_HEADERS` token to be configured for the _requestHeaderNames_ and/or _replyHeaderNames_ properties
(in addition to the existing `STANDARD_REQUEST_HEADERS` and `STANDARD_REPLY_HEADERS`) to map all user-defined headers.
Class `org.springframework.amqp.support.AmqpHeaders` identifies the default headers that will be used by the
`DefaultAmqpHeaderMapper`:
* amqp_appId
@@ -1102,6 +1111,8 @@ Class `org.springframework.amqp.support.AmqpHeaders` identifies the default head
* amqp_receivedDelay
* amqp_receivedDeliveryMode
* amqp_receivedExchange
* amqp_receivedRoutingKey
@@ -1144,6 +1155,10 @@ For example, to avoid copying the `x-received-from` header back to the reply you
before sending the reply to the AMQP Inbound Gateway.
Alternatively, you could explicitly list those properties that you actually want mapped instead of using
wildcards.
For these reasons, for inbound messages, the mapper by default does not map any `x-*` headers; it also does not map
the `deliveryMode` to `amqp_deliveryMode` header, to avoid propagation of that header from an inbound message to an
outbound message.
Instead, this header is mapped to `amqp_receivedDeliveryMode`, which is not mapped on output.
Starting with _version 4.3_, patterns in the header mappings can be negated by preceding the pattern with `!`.
Negated patterns get priority, so a list such as

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

@@ -133,10 +133,20 @@ See <<router>> for more information.
==== Header Mapping
===== General
AMQP, WS and XMPP header mappings (e.g. `request-header-mapping`, `reply-header-mapping`) now support negated
patterns.
See <<amqp-message-headers>>, <<ws-message-headers>>, and <<xmpp-message-headers>> for more information.
===== AMQP Header Mapping
Previously, only standard AMQP headers were mapped by default; users had to explicitly enable mapping of user-defined
headers.
With this release all headers are mapped by default.
In addition, the inbound `amqp_deliveryMode` header is no longer mapped by default.
See <<amqp-message-headers>> for more information.
==== Groovy Scripts
Groovy scripts can now be configured with the `compile-static` hint or any other `CompilerConfiguration` options.