From 470d6d880e4bd6fce8338185de2298dc6bab762b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 16 May 2018 13:49:38 -0400 Subject: [PATCH] INT-4397: Fix headers filtering for @Transformer (#2445) * INT-4397: Fix headers filtering for @Transformer JIRA: https://jira.spring.io/browse/INT-4397 The `AbstractMessageProcessingTransformer` doesn't honor a configured `notPropagatedHeaders` and copies all the request headers to the message to return * Add `setNotPropagatedHeaders()` into the `AbstractMessageProcessingTransformer` and implement there a logic to filter headers, similar to what we have in the `AbstractMessageProducingHandler` * Overrider `updateNotPropagatedHeaders()` in the `MessageTransformingHandler` to propagate `notPropagatedHeaders` to the `AbstractMessageProcessingTransformer` delegate * * Revert `final` for the `AbstractMessageProducingHandler.updateNotPropagatedHeaders()` * Override `addNotPropagatedHeaders()` for the `MessageTransformingHandler()` and populate `notPropagatedHeaders` into the target `AbstractMessageProcessingTransformer` from there * Also populate `notPropagatedHeaders` from the `AbstractMessageProcessingTransformer.doInit()` * Implement a `AbstractIntegrationMessageBuilder.filterAndCopyHeadersIfAbsent()` for a general logic to filter `notPropagatedHeaders` and copy the result headers set into the target message if they are absent * Use an new `filterAndCopyHeadersIfAbsent()` in the `AbstractMessageProducingHandler` and `AbstractMessageProcessingTransformer`to avoid code block duplication --- .../AbstractMessageProducingHandler.java | 17 +- .../AbstractIntegrationMessageBuilder.java | 243 ++++++++++-------- .../AbstractMessageProcessingTransformer.java | 42 ++- .../MessageTransformingHandler.java | 21 +- .../HeaderAnnotationTransformerTests.java | 26 +- 5 files changed, 217 insertions(+), 132 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 59961b0c8a..17c8fac401 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -19,7 +19,6 @@ package org.springframework.integration.handler; import java.util.Arrays; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -43,7 +42,6 @@ import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; -import org.springframework.util.PatternMatchUtils; import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; @@ -151,7 +149,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan headerPatterns.addAll(Arrays.asList(headers)); - this.notPropagatedHeaders = headerPatterns.toArray(new String[headerPatterns.size()]); + this.notPropagatedHeaders = headerPatterns.toArray(new String[0]); } boolean hasAsterisk = headerPatterns.contains("*"); @@ -388,17 +386,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan builder = this.getMessageBuilderFactory().withPayload(output); } if (!this.noHeadersPropagation && shouldCopyRequestHeaders()) { - if (this.selectiveHeaderPropagation) { - Map headersToCopy = new HashMap<>(requestHeaders); - - headersToCopy.entrySet() - .removeIf(entry -> PatternMatchUtils.simpleMatch(this.notPropagatedHeaders, entry.getKey())); - - builder.copyHeadersIfAbsent(headersToCopy); - } - else { - builder.copyHeadersIfAbsent(requestHeaders); - } + builder.filterAndCopyHeadersIfAbsent(requestHeaders, + this.selectiveHeaderPropagation ? this.notPropagatedHeaders : null); } return builder.build(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java index 4e431e0d23..548dc3c9f7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 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 java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Date; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -29,6 +30,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.PatternMatchUtils; /** * @author Gary Russell @@ -39,13 +42,141 @@ import org.springframework.util.Assert; */ public abstract class AbstractIntegrationMessageBuilder { + public AbstractIntegrationMessageBuilder setExpirationDate(Long expirationDate) { + return setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate); + } + + public AbstractIntegrationMessageBuilder setExpirationDate(Date expirationDate) { + if (expirationDate != null) { + return setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate.getTime()); + } + else { + return setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, null); + } + } + + public AbstractIntegrationMessageBuilder setCorrelationId(Object correlationId) { + return setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); + } + + public AbstractIntegrationMessageBuilder pushSequenceDetails(Object correlationId, int sequenceNumber, + int sequenceSize) { + + Object incomingCorrelationId = this.getCorrelationId(); + List> incomingSequenceDetails = getSequenceDetails(); + if (incomingCorrelationId != null) { + if (incomingSequenceDetails == null) { + incomingSequenceDetails = new ArrayList<>(); + } + else { + incomingSequenceDetails = new ArrayList<>(incomingSequenceDetails); + } + incomingSequenceDetails.add(Arrays.asList(incomingCorrelationId, + getSequenceNumber(), getSequenceSize())); + incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails); + } + if (incomingSequenceDetails != null) { + setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails); + } + return setCorrelationId(correlationId) + .setSequenceNumber(sequenceNumber) + .setSequenceSize(sequenceSize); + } + + public AbstractIntegrationMessageBuilder popSequenceDetails() { + List> incomingSequenceDetails = getSequenceDetails(); + if (incomingSequenceDetails == null) { + return this; + } + else { + incomingSequenceDetails = new ArrayList<>(incomingSequenceDetails); + } + List sequenceDetails = incomingSequenceDetails.remove(incomingSequenceDetails.size() - 1); + Assert.state(sequenceDetails.size() == 3, "Wrong sequence details (not created by MessageBuilder?): " + + sequenceDetails); + setCorrelationId(sequenceDetails.get(0)); + Integer sequenceNumber = (Integer) sequenceDetails.get(1); + Integer sequenceSize = (Integer) sequenceDetails.get(2); + if (sequenceNumber != null) { + setSequenceNumber(sequenceNumber); + } + if (sequenceSize != null) { + setSequenceSize(sequenceSize); + } + if (!incomingSequenceDetails.isEmpty()) { + setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails); + } + else { + removeHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS); + } + return this; + } + + public AbstractIntegrationMessageBuilder setReplyChannel(MessageChannel replyChannel) { + return setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel); + } + + public AbstractIntegrationMessageBuilder setReplyChannelName(String replyChannelName) { + return setHeader(MessageHeaders.REPLY_CHANNEL, replyChannelName); + } + + public AbstractIntegrationMessageBuilder setErrorChannel(MessageChannel errorChannel) { + return setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel); + } + + public AbstractIntegrationMessageBuilder setErrorChannelName(String errorChannelName) { + return setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName); + } + + public AbstractIntegrationMessageBuilder setSequenceNumber(Integer sequenceNumber) { + return setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); + } + + public AbstractIntegrationMessageBuilder setSequenceSize(Integer sequenceSize) { + return setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); + } + + public AbstractIntegrationMessageBuilder setPriority(Integer priority) { + return setHeader(IntegrationMessageHeaderAccessor.PRIORITY, priority); + } + + /** + * Remove headers from the provided map matching to the provided pattens + * and only after that copy the result into the target message headers. + * @param headersToCopy a map of headers to copy. + * @param headerPatternsToFilter an arrays of header patterns to filter before copying. + * @return the current {@link AbstractIntegrationMessageBuilder}. + * @since 5.1 + * @see #copyHeadersIfAbsent(Map) + */ + public AbstractIntegrationMessageBuilder filterAndCopyHeadersIfAbsent(Map headersToCopy, + String... headerPatternsToFilter) { + + Map headers = headersToCopy; + + if (!ObjectUtils.isEmpty(headerPatternsToFilter)) { + headers = new HashMap<>(headersToCopy); + headers.entrySet() + .removeIf(entry -> PatternMatchUtils.simpleMatch(headerPatternsToFilter, entry.getKey())); + } + + return copyHeadersIfAbsent(headers); + } + + protected abstract List> getSequenceDetails(); + + protected abstract Object getCorrelationId(); + + protected abstract Object getSequenceNumber(); + + protected abstract Object getSequenceSize(); + public abstract T getPayload(); public abstract Map getHeaders(); /** * Set the value for the given header name. If the provided value is null, the header will be removed. - * * @param headerName The header name. * @param headerValue The header value. * @return this. @@ -54,7 +185,6 @@ public abstract class AbstractIntegrationMessageBuilder { /** * Set the value for the given header name only if the header name is not already associated with a value. - * * @param headerName The header name. * @param headerValue The header value. * @return this. @@ -65,7 +195,6 @@ public abstract class AbstractIntegrationMessageBuilder { * Removes all headers provided via array of 'headerPatterns'. As the name suggests the array * may contain simple matching patterns for header names. Supported pattern styles are: * "xxx*", "*xxx", "*xxx*" and "xxx*yyy". - * * @param headerPatterns The header patterns. * @return this. */ @@ -82,10 +211,8 @@ public abstract class AbstractIntegrationMessageBuilder { * Copy the name-value pairs from the provided Map. This operation will overwrite any existing values. Use { * {@link #copyHeadersIfAbsent(Map)} to avoid overwriting values. Note that the 'id' and 'timestamp' header values * will never be overwritten. - * * @param headersToCopy The headers to copy. * @return this. - * * @see MessageHeaders#ID * @see MessageHeaders#TIMESTAMP */ @@ -93,115 +220,11 @@ public abstract class AbstractIntegrationMessageBuilder { /** * Copy the name-value pairs from the provided Map. This operation will not overwrite any existing values. - * * @param headersToCopy The headers to copy. * @return this. */ public abstract AbstractIntegrationMessageBuilder copyHeadersIfAbsent(@Nullable Map headersToCopy); - public AbstractIntegrationMessageBuilder setExpirationDate(Long expirationDate) { - return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate); - } - - public AbstractIntegrationMessageBuilder setExpirationDate(Date expirationDate) { - if (expirationDate != null) { - return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate.getTime()); - } - else { - return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, null); - } - } - - public AbstractIntegrationMessageBuilder setCorrelationId(Object correlationId) { - return this.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); - } - - public AbstractIntegrationMessageBuilder pushSequenceDetails(Object correlationId, int sequenceNumber, - int sequenceSize) { - Object incomingCorrelationId = this.getCorrelationId(); - List> incomingSequenceDetails = this.getSequenceDetails(); - if (incomingCorrelationId != null) { - if (incomingSequenceDetails == null) { - incomingSequenceDetails = new ArrayList>(); - } - else { - incomingSequenceDetails = new ArrayList>(incomingSequenceDetails); - } - incomingSequenceDetails.add(Arrays.asList(incomingCorrelationId, - this.getSequenceNumber(), this.getSequenceSize())); - incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails); - } - if (incomingSequenceDetails != null) { - this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails); - } - return setCorrelationId(correlationId).setSequenceNumber(sequenceNumber).setSequenceSize(sequenceSize); - } - - public AbstractIntegrationMessageBuilder popSequenceDetails() { - List> incomingSequenceDetails = this.getSequenceDetails(); - if (incomingSequenceDetails == null) { - return this; - } - else { - incomingSequenceDetails = new ArrayList>(incomingSequenceDetails); - } - List sequenceDetails = incomingSequenceDetails.remove(incomingSequenceDetails.size() - 1); - Assert.state(sequenceDetails.size() == 3, "Wrong sequence details (not created by MessageBuilder?): " - + sequenceDetails); - setCorrelationId(sequenceDetails.get(0)); - Integer sequenceNumber = (Integer) sequenceDetails.get(1); - Integer sequenceSize = (Integer) sequenceDetails.get(2); - if (sequenceNumber != null) { - setSequenceNumber(sequenceNumber); - } - if (sequenceSize != null) { - setSequenceSize(sequenceSize); - } - if (!incomingSequenceDetails.isEmpty()) { - this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails); - } - else { - this.removeHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS); - } - return this; - } - - protected abstract List> getSequenceDetails(); - - protected abstract Object getCorrelationId(); - - protected abstract Object getSequenceNumber(); - - protected abstract Object getSequenceSize(); - - public AbstractIntegrationMessageBuilder setReplyChannel(MessageChannel replyChannel) { - return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel); - } - - public AbstractIntegrationMessageBuilder setReplyChannelName(String replyChannelName) { - return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannelName); - } - - public AbstractIntegrationMessageBuilder setErrorChannel(MessageChannel errorChannel) { - return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel); - } - - public AbstractIntegrationMessageBuilder setErrorChannelName(String errorChannelName) { - return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName); - } - - public AbstractIntegrationMessageBuilder setSequenceNumber(Integer sequenceNumber) { - return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); - } - - public AbstractIntegrationMessageBuilder setSequenceSize(Integer sequenceSize) { - return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); - } - - public AbstractIntegrationMessageBuilder setPriority(Integer priority) { - return this.setHeader(IntegrationMessageHeaderAccessor.PRIORITY, priority); - } - public abstract Message build(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java index ed5f4b1bd1..2d6011af43 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2018 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,8 @@ package org.springframework.integration.transformer; +import java.util.Arrays; + import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; @@ -24,7 +26,9 @@ import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; /** * Base class for Message Transformers that delegate to a {@link MessageProcessor}. @@ -37,12 +41,16 @@ public abstract class AbstractMessageProcessingTransformer private final MessageProcessor messageProcessor; - private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); - - private volatile boolean messageBuilderFactorySet; - private BeanFactory beanFactory; + private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); + + private boolean messageBuilderFactorySet; + + private String[] notPropagatedHeaders; + + private boolean selectiveHeaderPropagation; + protected AbstractMessageProcessingTransformer(MessageProcessor messageProcessor) { Assert.notNull(messageProcessor, "messageProcessor must not be null"); this.messageProcessor = messageProcessor; @@ -85,6 +93,21 @@ public abstract class AbstractMessageProcessingTransformer return !(this.messageProcessor instanceof Lifecycle) || ((Lifecycle) this.messageProcessor).isRunning(); } + /** + * Set headers that will NOT be copied from the inbound message if + * the handler is configured to copy headers. + * @param headers the headers to not propagate from the inbound message. + * @since 5.1 + */ + public void setNotPropagatedHeaders(String... headers) { + if (!ObjectUtils.isEmpty(headers)) { + Assert.noNullElements(headers, "null elements are not allowed in 'headers'"); + this.notPropagatedHeaders = Arrays.copyOf(headers, headers.length); + } + + this.selectiveHeaderPropagation = !ObjectUtils.isEmpty(this.notPropagatedHeaders); + } + @Override public final Message transform(Message message) { Object result = this.messageProcessor.processMessage(message); @@ -94,7 +117,14 @@ public abstract class AbstractMessageProcessingTransformer if (result instanceof Message) { return (Message) result; } - return getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build(); + + MessageHeaders requestHeaders = message.getHeaders(); + + return getMessageBuilderFactory() + .withPayload(result) + .filterAndCopyHeadersIfAbsent(requestHeaders, + this.selectiveHeaderPropagation ? this.notPropagatedHeaders : null) + .build(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java index 65bb14f3fd..93d1f63fd9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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,8 @@ package org.springframework.integration.transformer; +import java.util.Collection; + import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -57,11 +59,28 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan ((NamedComponent) this.transformer).getComponentType() : "transformer"; } + @Override + public void addNotPropagatedHeaders(String... headers) { + super.addNotPropagatedHeaders(headers); + populateNotPropagatedHeadersIfAny(); + } + @Override protected void doInit() { if (this.getBeanFactory() != null && this.transformer instanceof BeanFactoryAware) { ((BeanFactoryAware) this.transformer).setBeanFactory(this.getBeanFactory()); } + + populateNotPropagatedHeadersIfAny(); + } + + private void populateNotPropagatedHeadersIfAny() { + Collection notPropagatedHeaders = getNotPropagatedHeaders(); + + if (this.transformer instanceof AbstractMessageProcessingTransformer && !notPropagatedHeaders.isEmpty()) { + ((AbstractMessageProcessingTransformer) this.transformer) + .setNotPropagatedHeaders(notPropagatedHeaders.toArray(new String[0])); + } } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/HeaderAnnotationTransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/HeaderAnnotationTransformerTests.java index 294dd81ce4..eadfe6d436 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/HeaderAnnotationTransformerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/HeaderAnnotationTransformerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2018 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,7 @@ package org.springframework.integration.handler; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.mockito.Mockito.mock; @@ -34,6 +35,8 @@ import org.springframework.messaging.handler.annotation.Header; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class HeaderAnnotationTransformerTests { @@ -103,6 +106,27 @@ public class HeaderAnnotationTransformerTests { } + @Test + public void testNotPropagatedHeaders() { + Object target = new TestTransformer(); + MethodInvokingTransformer transformer = new MethodInvokingTransformer(target, "evalFoo"); + MessageTransformingHandler handler = new MessageTransformingHandler(transformer); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setNotPropagatedHeaders(IntegrationMessageHeaderAccessor.CORRELATION_ID); + handler.afterPropertiesSet(); + QueueChannel outputChannel = new QueueChannel(); + handler.setOutputChannel(outputChannel); + handler.handleMessage( + MessageBuilder.withPayload("test") + .setCorrelationId("abc") + .setHeader("foo", "bar") + .build()); + Message result = outputChannel.receive(0); + assertNotNull(result); + assertEquals("BAR", result.getPayload()); + assertFalse(result.getHeaders().containsKey(IntegrationMessageHeaderAccessor.CORRELATION_ID)); + } + public static class TestTransformer { public String appendCorrelationId(Object payload,