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

INT-4096: Allow Configuration of ReadOnly Headers

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

* Introduce `spring.integration.readOnly.headers` Integration property
* Introduce `IntegrationMessageHeaderAccessor.setReadOnlyHeaders()` and use it from the overridden `IntegrationMessageHeaderAccessor.isReadOnly()`
* Add `DefaultMessageBuilderFactory.setReadOnlyHeaders()` and delegate the value to the new `MessageBuilder.readOnlyHeaders()`
* Modify `spring.integration.properties` for the `contentType` as a `readOnly` header for testing
* Ensure that provided logic works via an appropriate modification to the `ObjectToJsonTransformerParserTests`

Increase receive timeouts in the `OutboundGatewayFunctionTests`

Fix `FileInboundTransactionTests` for slow `WatchService` issue on Linux/OS X

Also redo the `tmp` dir logic to the `TemporaryFolder` `@ClassRule`

Fix unused `import` in the `FileInboundTransactionTests`

Polishing
This commit is contained in:
Artem Bilan
2016-09-12 20:48:34 -04:00
committed by Gary Russell
parent 9101b6a6ef
commit d1ed66835d
12 changed files with 121 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java
View File

@@ -17,12 +17,17 @@
package org.springframework.integration;
import java.io.Closeable;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
/**
*
@@ -53,10 +58,27 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
public static final String CLOSEABLE_RESOURCE = "closableResource";
private Set<String> readOnlyHeaders = new HashSet<String>();
public IntegrationMessageHeaderAccessor(Message<?> message) {
super(message);
}
/**
* Specify a list of headers which should be considered as read only
* and prohibited from being populated in the message.
* @param readOnlyHeaders the list of headers for {@code readOnly} mode.
* Defaults to {@link MessageHeaders#ID} and {@link MessageHeaders#TIMESTAMP}.
* @since 4.3.2
* @see #isReadOnly(String)
*/
public void setReadOnlyHeaders(String... readOnlyHeaders) {
Assert.noNullElements(readOnlyHeaders, "'readOnlyHeaders' must not be contain null items.");
if (!ObjectUtils.isEmpty(readOnlyHeaders)) {
this.readOnlyHeaders = new HashSet<String>(Arrays.asList(readOnlyHeaders));
}
}
public Long getExpirationDate() {
return this.getHeader(EXPIRATION_DATE, Long.class);
}
@@ -130,4 +152,9 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
}
}
@Override
protected boolean isReadOnly(String headerName) {
return super.isReadOnly(headerName) || this.readOnlyHeaders.contains(headerName);
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java
View File

@@ -427,7 +427,9 @@ public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, Bean
}
if (!alreadyRegistered) {
BeanDefinitionBuilder mbfBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultMessageBuilderFactory.class);
.genericBeanDefinition(DefaultMessageBuilderFactory.class)
.addPropertyValue("readOnlyHeaders",
IntegrationProperties.getExpressionFor(IntegrationProperties.READ_ONLY_HEADERS));
registry.registerBeanDefinition(
IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
mbfBuilder.getBeanDefinition());

8
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java
View File

@@ -72,10 +72,16 @@ public final class IntegrationProperties {
public static final String REQUIRE_COMPONENT_ANNOTATION = INTEGRATION_PROPERTIES_PREFIX + "messagingAnnotations.require.componentAnnotation";
/**
* Specifies the value of {@link org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor#requireComponentAnnotation}.
* Specifies the value of {@link org.springframework.integration.gateway.GatewayProxyFactoryBean#convertReceiveMessage}.
*/
public static final String GATEWAY_CONVERT_RECEIVE_MESSAGE = INTEGRATION_PROPERTIES_PREFIX + "messagingGateway.convertReceiveMessage";
/**
* Specifies the value of {@link org.springframework.integration.support.DefaultMessageBuilderFactory#readOnlyHeaders}.
*/
public static final String READ_ONLY_HEADERS = INTEGRATION_PROPERTIES_PREFIX + "readOnly.headers";
private static Properties defaults;
static {

21
spring-integration-core/src/main/java/org/springframework/integration/support/DefaultMessageBuilderFactory.java
View File

@@ -17,22 +17,39 @@
package org.springframework.integration.support;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
/**
* @author Gary Russell
* @author Artem Bilan
* @since 4.0
*
*/
public class DefaultMessageBuilderFactory implements MessageBuilderFactory {
private String[] readOnlyHeaders;
/**
* Specify a list of headers which should be considered as a read only
* and prohibited from the population to the message.
* @param readOnlyHeaders the list of headers for {@code readOnly} mode.
* Defaults to {@link MessageHeaders#ID} and {@link MessageHeaders#TIMESTAMP}.
* @since 4.3.2
*/
public void setReadOnlyHeaders(String... readOnlyHeaders) {
this.readOnlyHeaders = readOnlyHeaders;
}
@Override
public <T> MessageBuilder<T> fromMessage(Message<T> message) {
return MessageBuilder.fromMessage(message);
return MessageBuilder.fromMessage(message)
.readOnlyHeaders(this.readOnlyHeaders);
}
@Override
public <T> MessageBuilder<T> withPayload(T payload) {
return MessageBuilder.withPayload(payload);
return MessageBuilder.withPayload(payload)
.readOnlyHeaders(this.readOnlyHeaders);
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java
View File

@@ -274,6 +274,13 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
return this;
}
public MessageBuilder<T> readOnlyHeaders(String... readOnlyHeaders) {
this.headerAccessor.setReadOnlyHeaders(readOnlyHeaders);
return this;
}
@Override
@SuppressWarnings("unchecked")
public Message<T> build() {

2
spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties
View File

@@ -5,3 +5,5 @@ spring.integration.taskScheduler.poolSize=10
spring.integration.messagingTemplate.throwExceptionOnLateReply=false
spring.integration.messagingAnnotations.require.componentAnnotation=false
spring.integration.messagingGateway.convertReceiveMessage=false
# Defaults to MessageHeaders.ID and MessageHeaders.TIMESTAMP
spring.integration.readOnly.headers=

2
spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerParserTests-context.xml
View File

@@ -11,7 +11,7 @@
<object-to-json-transformer id="emptyContentTypeTransformer" input-channel="customObjectMapperInput" content-type=""/>
<object-to-json-transformer id="overridenContentTypeTransformer" input-channel="customObjectMapperInput" content-type="text/xml"/>
<object-to-json-transformer id="overriddenContentTypeTransformer" input-channel="customObjectMapperInput" content-type="text/xml"/>
<object-to-json-transformer id="customJsonObjectMapperTransformer" input-channel="customJsonObjectMapperInput"
object-mapper="customJsonObjectMapper"/>

41
spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerParserTests.java
View File

@@ -37,6 +37,7 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
import org.springframework.integration.support.json.JsonObjectMapperAdapter;
@@ -61,45 +62,57 @@ import com.fasterxml.jackson.databind.JsonNode;
public class ObjectToJsonTransformerParserTests {
@Autowired
private volatile ApplicationContext context;
private ApplicationContext context;
@Autowired
private volatile MessageChannel defaultObjectMapperInput;
private MessageChannel defaultObjectMapperInput;
@Autowired
private volatile MessageChannel customJsonObjectMapperInput;
private MessageChannel customJsonObjectMapperInput;
@Autowired
private volatile MessageChannel jsonNodeInput;
private MessageChannel jsonNodeInput;
@Autowired
private volatile MessageChannel boonJsonNodeInput;
private MessageChannel boonJsonNodeInput;
@Autowired
private DefaultMessageBuilderFactory defaultMessageBuilderFactory;
@Test
public void testContentType() {
ObjectToJsonTransformer transformer =
TestUtils.getPropertyValue(context.getBean("defaultTransformer"), "handler.transformer", ObjectToJsonTransformer.class);
TestUtils.getPropertyValue(context.getBean("defaultTransformer"), "handler.transformer",
ObjectToJsonTransformer.class);
assertEquals("application/json", TestUtils.getPropertyValue(transformer, "contentType"));
assertEquals(Jackson2JsonObjectMapper.class, TestUtils.getPropertyValue(transformer, "jsonObjectMapper").getClass());
assertEquals(Jackson2JsonObjectMapper.class,
TestUtils.getPropertyValue(transformer, "jsonObjectMapper").getClass());
Message<?> transformed = transformer.transform(MessageBuilder.withPayload("foo").build());
assertTrue(transformed.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE));
assertEquals("application/json", transformed.getHeaders().get(MessageHeaders.CONTENT_TYPE));
// spring.integration.readOnly.headers=contentType, so no 'contentType'
assertFalse(transformed.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE));
// Reset readOnlyHeaders to defaults. Therefore the 'contentType' should be presented in subsequent tests
this.defaultMessageBuilderFactory.setReadOnlyHeaders();
transformer =
TestUtils.getPropertyValue(context.getBean("emptyContentTypeTransformer"), "handler.transformer", ObjectToJsonTransformer.class);
TestUtils.getPropertyValue(context.getBean("emptyContentTypeTransformer"), "handler.transformer",
ObjectToJsonTransformer.class);
assertEquals("", TestUtils.getPropertyValue(transformer, "contentType"));
transformed = transformer.transform(MessageBuilder.withPayload("foo").build());
assertFalse(transformed.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE));
transformed = transformer.transform(MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "foo").build());
transformed = transformer.transform(MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE,
"foo").build());
assertNotNull(transformed.getHeaders().get(MessageHeaders.CONTENT_TYPE));
assertEquals("foo", transformed.getHeaders().get(MessageHeaders.CONTENT_TYPE));
transformer =
TestUtils.getPropertyValue(context.getBean("overridenContentTypeTransformer"), "handler.transformer", ObjectToJsonTransformer.class);
TestUtils.getPropertyValue(context.getBean("overriddenContentTypeTransformer"), "handler.transformer",
ObjectToJsonTransformer.class);
assertEquals("text/xml", TestUtils.getPropertyValue(transformer, "contentType"));
}
@@ -165,7 +178,8 @@ public class ObjectToJsonTransformerParserTests {
assertThat(payload, Matchers.instanceOf(JsonNode.class));
StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
evaluationContext.addPropertyAccessor(new JsonPropertyAccessor());
Expression expression = new SpelExpressionParser().parseExpression("firstName.toString() == 'John' and age.toString() == '42'");
Expression expression = new SpelExpressionParser()
.parseExpression("firstName.toString() == 'John' and age.toString() == '42'");
assertTrue(expression.getValue(evaluationContext, payload, Boolean.class));
}
@@ -195,6 +209,7 @@ public class ObjectToJsonTransformerParserTests {
public String toJson(Object value) throws Exception {
return "{" + value.toString() + "}";
}
}
}

1
spring-integration-core/src/test/resources/META-INF/spring.integration.properties
View File

@@ -3,3 +3,4 @@
#spring.integration.channels.maxBroadcastSubscribers=1
spring.integration.taskScheduler.poolSize=20
spring.integration.messagingTemplate.throwExceptionOnLateReply=true
spring.integration.readOnly.headers=contentType

5
spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml
View File

@@ -12,7 +12,8 @@
<context:property-placeholder/>
<int-file:inbound-channel-adapter id="pseudoTx"
channel="input" auto-startup="false" directory="${java.io.tmpdir}/si-test1"
channel="input" auto-startup="false"
directory="#{T (org.springframework.integration.file.FileInboundTransactionTests).tmpDir.root}/si-test1"
use-watch-service="true">
<int:poller fixed-rate="500">
<int:transactional synchronization-factory="syncFactoryA"/>
@@ -32,7 +33,7 @@
<int:channel id="txInput" />
<int-file:inbound-channel-adapter id="realTx" channel="txInput" auto-startup="false"
directory="${java.io.tmpdir}/si-test2">
directory="#{T (org.springframework.integration.file.FileInboundTransactionTests).tmpDir.root}/si-test2">
<int:poller fixed-rate="500">
<int:transactional transaction-manager="txManager" synchronization-factory="syncFactoryB"/>
</int:poller>

30
spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java
View File

@@ -27,11 +27,12 @@ import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
@@ -39,6 +40,7 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.TransactionDefinition;
@@ -54,8 +56,12 @@ import org.springframework.transaction.support.DefaultTransactionStatus;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class FileInboundTransactionTests {
@ClassRule
public static TemporaryFolder tmpDir = new TemporaryFolder();
@Autowired
private SourcePollingChannelAdapter pseudoTx;
@@ -77,9 +83,6 @@ public class FileInboundTransactionTests {
@Autowired
private DummyTxManager transactionManager;
@Value("${java.io.tmpdir}")
private String tmpDir;
@Test
public void testNoTx() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
@@ -95,17 +98,16 @@ public class FileInboundTransactionTests {
}
});
pseudoTx.start();
new File(tmpDir + "/si-test1").mkdir();
File file = new File(tmpDir + "/si-test1/foo");
File file = new File(tmpDir.getRoot(), "si-test1/foo");
file.createNewFile();
Message<?> result = successChannel.receive(40000);
Message<?> result = successChannel.receive(60000);
assertNotNull(result);
assertEquals(Boolean.TRUE, result.getPayload());
assertFalse(file.delete());
crash.set(true);
file = new File(tmpDir + "/si-test1/bar");
file = new File(tmpDir.getRoot(), "si-test1/bar");
file.createNewFile();
result = failureChannel.receive(10000);
result = failureChannel.receive(60000);
assertNotNull(result);
assertTrue(file.delete());
assertEquals("foo", result.getPayload());
@@ -132,18 +134,17 @@ public class FileInboundTransactionTests {
}
});
realTx.start();
new File(tmpDir + "/si-test2").mkdir();
File file = new File(tmpDir + "/si-test2/baz");
File file = new File(tmpDir.getRoot(), "si-test2/baz");
file.createNewFile();
Message<?> result = successChannel.receive(40000);
Message<?> result = successChannel.receive(60000);
assertNotNull(result);
assertEquals(Boolean.TRUE, result.getPayload());
assertTrue(file.delete());
assertTrue(transactionManager.getCommitted());
crash.set(true);
file = new File(tmpDir + "/si-test2/qux");
file = new File(tmpDir.getRoot(), "si-test2/qux");
file.createNewFile();
result = failureChannel.receive(10000);
result = failureChannel.receive(60000);
assertNotNull(result);
assertTrue(file.delete());
assertEquals(Boolean.TRUE, result.getPayload());
@@ -193,6 +194,7 @@ public class FileInboundTransactionTests {
public boolean getRolledBack() {
return rolledBack;
}
}
}

14
spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.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.
@@ -114,7 +114,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(connectionFactory);
template.setReceiveTimeout(5000);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
@@ -166,7 +166,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(5000);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue2);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
@@ -220,7 +220,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(5000);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue3);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
@@ -272,7 +272,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(5000);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue4);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
@@ -326,7 +326,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(5000);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue5);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
@@ -378,7 +378,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(5000);
template.setReceiveTimeout(10000);
javax.jms.Message request = template.receive(requestQueue6);
assertNotNull(request);
final javax.jms.Message jmsReply = request;