diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java index bb7badf64..faebb489e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java @@ -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 trackHistoryProperties = StringUtils.commaDelimitedListToSet(bindingProperties.getTrackedProperties()); + Map channelBindingServicePropertiesMap = channelBindingServiceProperties.asMapProperties(); + final Map 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 map = new LinkedHashMap(); 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) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 4c50628dd..fb2feb648 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -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); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index e1d821030..0e6c66a83 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java index cbf208270..b7c4b7957 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java @@ -232,6 +232,8 @@ public class ChannelBindingServiceProperties { Map 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 entry : getBindings().entrySet()) { properties.put(entry.getKey(), entry.getValue().toString()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurerTests.java new file mode 100644 index 000000000..b4b93a2cd --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurerTests.java @@ -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 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> headerValues = (List>) message.getHeaders().get(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER); + Map 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 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> headerValues = (List>) message.getHeaders().get(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER); + Map 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 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> headerValues = (List>) message.getHeaders().get(MessageHistoryTrackerConfigurer.HISTORY_TRACKING_HEADER); + Map 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()); + } +}