diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java index 9f5cfa2721..f373343748 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/DefaultJmsHeaderMapper.java @@ -144,21 +144,28 @@ public class DefaultJmsHeaderMapper implements JmsHeaderMapper { if (StringUtils.hasText(headerName) && !headerName.startsWith(JmsHeaders.PREFIX) && jmsMessage.getObjectProperty(headerName) == null) { Object value = entry.getValue(); - if (value != null && SUPPORTED_PROPERTY_TYPES.contains(value.getClass())) { - try { - String propertyName = this.fromHeaderName(headerName); - jmsMessage.setObjectProperty(propertyName, value); - } - catch (Exception e) { - if (headerName.startsWith("JMSX") - || headerName.equals(IntegrationMessageHeaderAccessor.PRIORITY)) { - if (logger.isTraceEnabled()) { - logger.trace("skipping reserved header, it cannot be set by client: " + headerName); + if (value != null) { + if (SUPPORTED_PROPERTY_TYPES.contains(value.getClass())) { + try { + String propertyName = this.fromHeaderName(headerName); + jmsMessage.setObjectProperty(propertyName, value); + } + catch (Exception e) { + if (headerName.startsWith("JMSX") + || headerName.equals(IntegrationMessageHeaderAccessor.PRIORITY)) { + if (logger.isTraceEnabled()) { + logger.trace("skipping reserved header, it cannot be set by client: " + + headerName); + } + } + else if (logger.isWarnEnabled()) { + logger.warn("failed to map Message header '" + headerName + "' to JMS property", e); } } - else if (logger.isWarnEnabled()) { - logger.warn("failed to map Message header '" + headerName + "' to JMS property", e); - } + } + else if (IntegrationMessageHeaderAccessor.CORRELATION_ID.equals(headerName)) { + String propertyName = fromHeaderName(headerName); + jmsMessage.setObjectProperty(propertyName, value.toString()); } } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/ActiveMQMultiContextTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/ActiveMQMultiContextTests.java index 28cefde979..eaf2c00b3b 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/ActiveMQMultiContextTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/ActiveMQMultiContextTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -41,6 +41,7 @@ public abstract class ActiveMQMultiContextTests { @BeforeClass public static void startUp() throws Exception { + connectionFactory.setCacheConsumers(false); connectionFactory.createConnection().close(); } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SplitterAggregatorTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SplitterAggregatorTests.java new file mode 100644 index 0000000000..e5c8b15b12 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SplitterAggregatorTests.java @@ -0,0 +1,129 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.jms; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.splitter.DefaultMessageSplitter; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 4.3 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class SplitterAggregatorTests extends ActiveMQMultiContextTests { + + @Autowired + private MessageChannel splitChannel; + + @Autowired + private PollableChannel resultChannel; + + @SuppressWarnings("unchecked") + @Test + public void testSplitterAggregatorOverJms() { + List payload = Arrays.asList(1, 2, 3, 4, 5, 6); + this.splitChannel.send(new GenericMessage>(payload)); + Message message = this.resultChannel.receive(10000); + assertNotNull(message); + Collections.sort(((List) message.getPayload())); + assertEquals(payload, message.getPayload()); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + @Splitter(inputChannel = "splitChannel") + public MessageHandler splitter() { + DefaultMessageSplitter splitter = new DefaultMessageSplitter(); + splitter.setOutputChannelName("toJmsChannel"); + return splitter; + } + + @Bean + @ServiceActivator(inputChannel = "toJmsChannel") + public MessageHandler toJms() { + JmsSendingMessageHandler handler = new JmsSendingMessageHandler(new JmsTemplate(connectionFactory)); + handler.setDestinationName("splitterAggregator"); + return handler; + } + + @Bean + @InboundChannelAdapter(value = "aggregateChannel", + poller = @Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")) + public MessageSource fromJms() { + JmsDestinationPollingSource source = new JmsDestinationPollingSource(new JmsTemplate(connectionFactory)); + source.setDestinationName("splitterAggregator"); + return source; + } + + @Bean + public MessageChannel aggregateChannel() { + return new DirectChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "aggregateChannel") + public MessageHandler aggregator() { + AggregatingMessageHandler handler = + new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()); + handler.setOutputChannel(resultChannel()); + return handler; + } + + @Bean + public PollableChannel resultChannel() { + return new QueueChannel(); + } + + } + +} diff --git a/src/reference/asciidoc/jms.adoc b/src/reference/asciidoc/jms.adoc index 6e6fe159e4..62fec88d5a 100644 --- a/src/reference/asciidoc/jms.adoc +++ b/src/reference/asciidoc/jms.adoc @@ -559,6 +559,14 @@ Custom header mapper could also be provided via `header-mapper` attribute of inb IMPORTANT: Since _version 4.0_, the `JMSPriority` header is mapped to the standard `priority` header for inbound messages (previously, the `priority` header was only used for outbound messages). To revert to the previous behavior (do not map inbound priority), use the `mapInboundPriority` property of `DefaultJmsHeaderMapper` with argument set to `false`. +IMPORTANT: Since _version 4.3_, the `DefaultJmsHeaderMapper` now maps the standard `correlationId` header as a message +property by invoking its `toString()` method (`correlationId` is often a `UUID`, which is not a type that is supported +by JMS). +On the inbound side, it is mapped as a `String`. +This is independent of the `jms_correlationId` header which is mapped to/from the `JMSCorrelationID` header. +The `JMSCorrelationID` is generally used to correlate requests and replies whereas the `correlationId` is often used +to combine related messages into a group (such as with an aggregator or resequencer). + [[jms-conversion-and-marshalling]] === Message Conversion, Marshalling and Unmarshalling diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 335611ac11..da43e0b3f0 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -26,3 +26,9 @@ If you have such configuration, simply remove the `reply-channel`. The customizable `userFlag` added in 4.2.2 to provide customization of the flag used to denote that the mail has been seen is now available using the XML namespace. See <> for more information. + +==== JMS Changes + +The `DefaultJmsHeaderMapper` now maps the standard `correlationId` header as a message property by invoking its +`toString()` method. +See <> for more information.