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

INT-4397: Fix headers filtering for @Transformer (#2445)

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2018-05-16 13:49:38 -04:00
committed by Gary Russell
parent 59d6c279a4
commit 470d6d880e
5 changed files with 217 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -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<String, Object> 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();
}

243
spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java
View File

@@ -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<T> {
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Long expirationDate) {
return setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate);
}
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Date expirationDate) {
if (expirationDate != null) {
return setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate.getTime());
}
else {
return setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, null);
}
}
public AbstractIntegrationMessageBuilder<T> setCorrelationId(Object correlationId) {
return setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
}
public AbstractIntegrationMessageBuilder<T> pushSequenceDetails(Object correlationId, int sequenceNumber,
int sequenceSize) {
Object incomingCorrelationId = this.getCorrelationId();
List<List<Object>> 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<T> popSequenceDetails() {
List<List<Object>> incomingSequenceDetails = getSequenceDetails();
if (incomingSequenceDetails == null) {
return this;
}
else {
incomingSequenceDetails = new ArrayList<>(incomingSequenceDetails);
}
List<Object> 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<T> setReplyChannel(MessageChannel replyChannel) {
return setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel);
}
public AbstractIntegrationMessageBuilder<T> setReplyChannelName(String replyChannelName) {
return setHeader(MessageHeaders.REPLY_CHANNEL, replyChannelName);
}
public AbstractIntegrationMessageBuilder<T> setErrorChannel(MessageChannel errorChannel) {
return setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel);
}
public AbstractIntegrationMessageBuilder<T> setErrorChannelName(String errorChannelName) {
return setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName);
}
public AbstractIntegrationMessageBuilder<T> setSequenceNumber(Integer sequenceNumber) {
return setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
}
public AbstractIntegrationMessageBuilder<T> setSequenceSize(Integer sequenceSize) {
return setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
}
public AbstractIntegrationMessageBuilder<T> 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<T> filterAndCopyHeadersIfAbsent(Map<String, ?> headersToCopy,
String... headerPatternsToFilter) {
Map<String, ?> 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<List<Object>> getSequenceDetails();
protected abstract Object getCorrelationId();
protected abstract Object getSequenceNumber();
protected abstract Object getSequenceSize();
public abstract T getPayload();
public abstract Map<String, Object> getHeaders();
/**
* Set the value for the given header name. If the provided value is <code>null</code>, 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<T> {
/**
* 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<T> {
* 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<T> {
* 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<T> {
/**
* Copy the name-value pairs from the provided Map. This operation will <em>not</em> overwrite any existing values.
*
* @param headersToCopy The headers to copy.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(@Nullable Map<String, ?> headersToCopy);
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Long expirationDate) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate);
}
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Date expirationDate) {
if (expirationDate != null) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate.getTime());
}
else {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, null);
}
}
public AbstractIntegrationMessageBuilder<T> setCorrelationId(Object correlationId) {
return this.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
}
public AbstractIntegrationMessageBuilder<T> pushSequenceDetails(Object correlationId, int sequenceNumber,
int sequenceSize) {
Object incomingCorrelationId = this.getCorrelationId();
List<List<Object>> incomingSequenceDetails = this.getSequenceDetails();
if (incomingCorrelationId != null) {
if (incomingSequenceDetails == null) {
incomingSequenceDetails = new ArrayList<List<Object>>();
}
else {
incomingSequenceDetails = new ArrayList<List<Object>>(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<T> popSequenceDetails() {
List<List<Object>> incomingSequenceDetails = this.getSequenceDetails();
if (incomingSequenceDetails == null) {
return this;
}
else {
incomingSequenceDetails = new ArrayList<List<Object>>(incomingSequenceDetails);
}
List<Object> 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<List<Object>> getSequenceDetails();
protected abstract Object getCorrelationId();
protected abstract Object getSequenceNumber();
protected abstract Object getSequenceSize();
public AbstractIntegrationMessageBuilder<T> setReplyChannel(MessageChannel replyChannel) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel);
}
public AbstractIntegrationMessageBuilder<T> setReplyChannelName(String replyChannelName) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannelName);
}
public AbstractIntegrationMessageBuilder<T> setErrorChannel(MessageChannel errorChannel) {
return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel);
}
public AbstractIntegrationMessageBuilder<T> setErrorChannelName(String errorChannelName) {
return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName);
}
public AbstractIntegrationMessageBuilder<T> setSequenceNumber(Integer sequenceNumber) {
return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
}
public AbstractIntegrationMessageBuilder<T> setSequenceSize(Integer sequenceSize) {
return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
}
public AbstractIntegrationMessageBuilder<T> setPriority(Integer priority) {
return this.setHeader(IntegrationMessageHeaderAccessor.PRIORITY, priority);
}
public abstract Message<T> build();
}

42
spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java
View File

@@ -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();
}
}

21
spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

@@ -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<String> notPropagatedHeaders = getNotPropagatedHeaders();
if (this.transformer instanceof AbstractMessageProcessingTransformer && !notPropagatedHeaders.isEmpty()) {
((AbstractMessageProcessingTransformer) this.transformer)
.setNotPropagatedHeaders(notPropagatedHeaders.toArray(new String[0]));
}
}
@Override

26
spring-integration-core/src/test/java/org/springframework/integration/handler/HeaderAnnotationTransformerTests.java
View File

@@ -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,