diff --git a/build.gradle b/build.gradle index 524b0d2bdd..4e1a8311f2 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java index d967ef0005..fa62c21321 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.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. @@ -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"); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java index 7a2dac1fc1..6f84a66db1 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -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); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java index de0a1590eb..8443244850 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java @@ -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; diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 0d02c49ec0..82c4c0e619 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.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. @@ -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) { diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index bbbe3e2a28..c31e8f6c36 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.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. @@ -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; 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 80a0f1edae..abc7900c75 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 @@ -77,7 +77,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler private volatile ExpressionEvaluatingMessageProcessor exchangeNameGenerator; - private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); private volatile Expression confirmCorrelationExpression; 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 0fa2b34d7b..d6d019a128 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 @@ -77,6 +77,7 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper 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 - - - - - - - - - + @@ -46,5 +50,5 @@ - + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java index 22017d5745..a338866822 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.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. @@ -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 " + 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 da61bcb811..a3d72277a6 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 @@ -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(); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index b067de7d3e..a7557f9c88 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -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); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java index b74760964d..caf5a4483f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.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. @@ -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 amqpMessage = 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 d0b7a6006a..2b5cf3c4b6 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 @@ -51,7 +51,7 @@ public class DefaultAmqpHeaderMapperTests { @Test public void fromHeaders() { - DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); Map headerMap = new HashMap(); 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 headerMap = new HashMap(); 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 headerMap = new HashMap(); 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 headerMap = new HashMap(); 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 headerMap = new HashMap(); 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 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")); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java index 02a01090b8..b54d321285 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java @@ -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. diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 2f2af7cacf..fa554d2c71 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 71289557cc..beda975ad9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -133,10 +133,20 @@ See <> 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 <>, <>, and <> 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 <> for more information. + ==== Groovy Scripts Groovy scripts can now be configured with the `compile-static` hint or any other `CompilerConfiguration` options.