SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make history tracker properties configurable

Browse Source 
- Add property `spring.cloud.stream.bindings.<channelName>.trackHistoryProperties` with comma separated string values that specify the exact
property names in `ChannelBindingServiceProperties` to be added in the message tracker header
 - Add tests

This resolves #252

Add test for empty trackedHistories

 - rename `trackHistoryProperties` to `trackedProperties`
 - updated Javadoc
This commit is contained in:
Ilayaperumal Gopinathan
2016-02-10 17:43:17 +05:30
committed by Marius Bogoevici
parent 047a3f8d4f
commit 81e6769b74
5 changed files with 179 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-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.
@@ -17,10 +17,11 @@ package org.springframework.cloud.stream.binding;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.integration.channel.ChannelInterceptorAware;
@@ -28,6 +29,7 @@ import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.util.StringUtils;
/**
* Class that is responsible for configuring the message channel to enable message track history.
@@ -40,17 +42,31 @@ public class MessageHistoryTrackerConfigurer implements MessageChannelConfigurer
private final ChannelBindingServiceProperties channelBindingServiceProperties;
@Autowired
MessageBuilderFactory messageBuilderFactory;
private final MessageBuilderFactory messageBuilderFactory;
public MessageHistoryTrackerConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties) {
public MessageHistoryTrackerConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties,
MessageBuilderFactory messageBuilderFactory) {
this.channelBindingServiceProperties = channelBindingServiceProperties;
this.messageBuilderFactory = messageBuilderFactory;
}
@Override
public void configureMessageChannel(MessageChannel messageChannel, String channelName) {
BindingProperties bindingProperties = channelBindingServiceProperties.getBindings().get(channelName);
if (bindingProperties != null && Boolean.TRUE.equals(bindingProperties.isTrackHistory())) {
final Set<String> trackHistoryProperties = StringUtils.commaDelimitedListToSet(bindingProperties.getTrackedProperties());
Map<String, Object> channelBindingServicePropertiesMap = channelBindingServiceProperties.asMapProperties();
final Map<String, Object> historyMap = new HashMap<>();
if (bindingProperties.getTrackedProperties().equalsIgnoreCase("all")) {
historyMap.putAll(channelBindingServicePropertiesMap);
}
else {
for (String property : trackHistoryProperties) {
if (channelBindingServicePropertiesMap.keySet().contains(property)) {
historyMap.put(property, channelBindingServicePropertiesMap.get(property));
}
}
}
if (messageChannel instanceof ChannelInterceptorAware) {
((ChannelInterceptorAware) messageChannel).addInterceptor(new ChannelInterceptorAdapter() {
@@ -67,7 +83,8 @@ public class MessageHistoryTrackerConfigurer implements MessageChannelConfigurer
}
Map<String, Object> map = new LinkedHashMap<String, Object>();
map.put("thread", Thread.currentThread().getName());
history.add(channelBindingServiceProperties.asMapProperties());
map.putAll(historyMap);
history.add(map);
Message<?> out = messageBuilderFactory
.fromMessage(message)
.setHeader(HISTORY_TRACKING_HEADER, history)

18
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java
View File

@@ -53,8 +53,18 @@ public class BindingProperties {
private String binder;
/**
* Flag to indicate if the message header needs to be updated with the trackedProperties.
*/
private Boolean trackHistory;
/**
* Comma separated list of binding properties to track.
* By default the properties such as the current thread name and 'timestamp' are added if the 'trackHistory` is
* enabled.
*/
private String trackedProperties = "all";
// Outbound properties
private String requiredGroups;
@@ -253,6 +263,14 @@ public class BindingProperties {
this.requiredGroups = requiredGroups;
}
public String getTrackedProperties() {
return this.trackedProperties;
}
public void setTrackedProperties(String trackedProperties) {
this.trackedProperties = trackedProperties;
}
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("destination=" + this.destination);

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java
View File

@@ -23,6 +23,7 @@ import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -50,6 +51,7 @@ import org.springframework.expression.PropertyAccessor;
import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.json.JsonPropertyAccessor;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.core.DestinationResolver;
@@ -67,6 +69,9 @@ import org.springframework.messaging.core.DestinationResolver;
@EnableConfigurationProperties(ChannelBindingServiceProperties.class)
public class ChannelBindingServiceConfiguration {
@Autowired
MessageBuilderFactory messageBuilderFactory;
@Bean
// This conditional is intentionally not in an autoconfig (usually a bad idea) because
// it is used to detect a ChannelBindingService in the parent context (which we know
@@ -92,7 +97,7 @@ public class ChannelBindingServiceConfiguration {
@Bean
public MessageHistoryTrackerConfigurer messageHistoryTrackerConfigurer
(ChannelBindingServiceProperties channelBindingServiceProperties) {
return new MessageHistoryTrackerConfigurer(channelBindingServiceProperties);
return new MessageHistoryTrackerConfigurer(channelBindingServiceProperties, messageBuilderFactory);
}
@Bean

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java
View File

@@ -232,6 +232,8 @@ public class ChannelBindingServiceProperties {
Map<String, Object> properties = new HashMap<>();
properties.put("instanceIndex", String.valueOf(getInstanceIndex()));
properties.put("instanceCount", String.valueOf(getInstanceCount()));
properties.put("defaultBinder", getDefaultBinder());
properties.put("dynamicDestinations", getDynamicDestinations());
// Add Bindings properties
for (Map.Entry<String, BindingProperties> entry : getBindings().entrySet()) {
properties.put(entry.getKey(), entry.getValue().toString());

130
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurerTests.java Normal file
View File

@@ -0,0 +1,130 @@
/*
* Copyright 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.
* 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.cloud.stream.binding;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Test;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.MutableMessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
* @author Ilayaperumal Gopinathan
*/
public class MessageHistoryTrackerConfigurerTests {
@Test
public void testHistoryTrackAll() {
ChannelBindingServiceProperties serviceProperties = new ChannelBindingServiceProperties();
serviceProperties.setInstanceCount(2);
serviceProperties.setInstanceIndex(0);
Map<String, BindingProperties> bindingPropertiesMap = new HashMap<>();
BindingProperties bindingProperties = new BindingProperties();
bindingProperties.setTrackHistory(true);
bindingPropertiesMap.put("input", bindingProperties);
bindingPropertiesMap.put("test1", new BindingProperties());
serviceProperties.setBindings(bindingPropertiesMap);
MessageHistoryTrackerConfigurer historyTrackerConfigurer = new MessageHistoryTrackerConfigurer(serviceProperties,
new MutableMessageBuilderFactory());
DirectChannel messageChannel = new DirectChannel();
messageChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Assert.isTrue(message.getHeaders().containsKey(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER));
List<Map<String, Object>> headerValues = (List<Map<String, Object>>) message.getHeaders().get(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER);
Map<String, Object> historyValues = headerValues.get(0);
Assert.isTrue(historyValues.containsKey("instanceIndex") && historyValues.get("instanceIndex").equals("0"), "Instance index must exist with value '0'");
Assert.isTrue(historyValues.containsKey("instanceCount") && historyValues.get("instanceCount").equals("2"), "Instance count must exist with value '2'");
Assert.isTrue(historyValues.containsKey("input"), "Binding properties must exist for the channel 'input'");
Assert.isTrue(historyValues.containsKey("test1"), "Binding properties must exist for the channel 'test1'");
}
});
historyTrackerConfigurer.configureMessageChannel(messageChannel, "input");
messageChannel.send(MessageBuilder.withPayload("test").build());
}
@Test
public void testHistoryTrackSpecificProperties() {
ChannelBindingServiceProperties serviceProperties = new ChannelBindingServiceProperties();
serviceProperties.setInstanceCount(2);
serviceProperties.setInstanceIndex(0);
Map<String, BindingProperties> bindingPropertiesMap = new HashMap<>();
BindingProperties bindingProperties = new BindingProperties();
bindingProperties.setTrackHistory(true);
bindingProperties.setTrackedProperties("input,instanceIndex");
bindingPropertiesMap.put("input", bindingProperties);
bindingPropertiesMap.put("test1", new BindingProperties());
serviceProperties.setBindings(bindingPropertiesMap);
MessageHistoryTrackerConfigurer historyTrackerConfigurer = new MessageHistoryTrackerConfigurer(serviceProperties,
new MutableMessageBuilderFactory());
DirectChannel messageChannel = new DirectChannel();
messageChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Assert.isTrue(message.getHeaders().containsKey(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER));
List<Map<String, Object>> headerValues = (List<Map<String, Object>>) message.getHeaders().get(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER);
Map<String, Object> historyValues = headerValues.get(0);
Assert.isTrue(historyValues.containsKey("thread"), "Default property 'thread' should exist.");
Assert.isTrue(historyValues.containsKey("timestamp"), "Default property 'timestamp' should exist.");
Assert.isTrue(historyValues.containsKey("instanceIndex") && historyValues.get("instanceIndex").equals("0"), "Instance index must exist with value '0'");
Assert.isTrue(!historyValues.containsKey("instanceCount"), "Instance count should not be in the tracker header");
Assert.isTrue(historyValues.containsKey("input"), "Binding properties must exist for the channel 'input'");
Assert.isTrue(!historyValues.containsKey("test1"), "Binding properties for the channel 'test1' should not be in the tracker header");
}
});
historyTrackerConfigurer.configureMessageChannel(messageChannel, "input");
messageChannel.send(MessageBuilder.withPayload("test").build());
}
@Test
public void testHistoryTrackEmptyProperties() {
ChannelBindingServiceProperties serviceProperties = new ChannelBindingServiceProperties();
serviceProperties.setInstanceCount(2);
serviceProperties.setInstanceIndex(0);
Map<String, BindingProperties> bindingPropertiesMap = new HashMap<>();
BindingProperties bindingProperties = new BindingProperties();
bindingProperties.setTrackHistory(true);
bindingProperties.setTrackedProperties("");
bindingPropertiesMap.put("input", bindingProperties);
bindingPropertiesMap.put("test1", new BindingProperties());
serviceProperties.setBindings(bindingPropertiesMap);
MessageHistoryTrackerConfigurer historyTrackerConfigurer = new MessageHistoryTrackerConfigurer(serviceProperties,
new MutableMessageBuilderFactory());
DirectChannel messageChannel = new DirectChannel();
messageChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Assert.isTrue(message.getHeaders().containsKey(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER));
List<Map<String, Object>> headerValues = (List<Map<String, Object>>) message.getHeaders().get(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER);
Map<String, Object> historyValues = headerValues.get(0);
Assert.isTrue(historyValues.containsKey("thread"), "Default property 'thread' should exist.");
Assert.isTrue(historyValues.containsKey("timestamp"), "Default property 'timestamp' should exist.");
}
});
historyTrackerConfigurer.configureMessageChannel(messageChannel, "input");
messageChannel.send(MessageBuilder.withPayload("test").build());
}
}