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

Removing message history support

Browse Source
This commit is contained in:
Marius Bogoevici
2016-03-17 18:11:34 -04:00
committed by Ilayaperumal Gopinathan
parent 216149b78d
commit 95b0197dd7
6 changed files with 3 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java
View File

@@ -19,8 +19,6 @@ import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -68,32 +66,7 @@ public class MessageChannelConfigurerTests {
assertTrue(latch.await(10, TimeUnit.SECONDS));
testSink.input().unsubscribe(messageHandler);
}
@Test
public void testHistoryTrackerConfigurer() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
MessageHandler messageHandler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue("Message header should have tracking history info",
message.getHeaders().containsKey("SPRING_CLOUD_STREAM_HISTORY"));
@SuppressWarnings("unchecked")
Map<String, String> headerValue = ((Map<String, String>) ((List<?>) message.getHeaders()
.get("SPRING_CLOUD_STREAM_HISTORY")).get(0));
String inputBindingProps = headerValue.get("input");
assertTrue(inputBindingProps.contains("destination=configure"));
assertTrue(inputBindingProps.contains("trackHistory=true"));
assertTrue(headerValue.get("instanceIndex").equals("0"));
assertTrue(headerValue.get("instanceCount").equals("1"));
latch.countDown();
}
};
testSink.input().subscribe(messageHandler);
testSink.input().send(MessageBuilder.withPayload("{\"test\":\"value\"}").build());
assertTrue(latch.await(10, TimeUnit.SECONDS));
testSink.input().unsubscribe(messageHandler);
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties")

2
spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties
View File

@@ -1,4 +1,2 @@
spring.cloud.stream.bindings.input.destination=configure1
spring.cloud.stream.bindings.input.contentType=application/x-spring-tuple
spring.cloud.stream.bindings.input.trackHistory=true
spring.cloud.stream.bindings.input.concurrency=1

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

@@ -1,99 +0,0 @@
/*
* 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.
* 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.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.integration.channel.ChannelInterceptorAware;
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.
*
* @author Ilayaperumal Gopinathan
*/
public class MessageHistoryTrackerConfigurer implements MessageChannelConfigurer {
public static final String HISTORY_TRACKING_HEADER = "SPRING_CLOUD_STREAM_HISTORY";
private final ChannelBindingServiceProperties channelBindingServiceProperties;
private final MessageBuilderFactory messageBuilderFactory;
public MessageHistoryTrackerConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties,
MessageBuilderFactory messageBuilderFactory) {
this.channelBindingServiceProperties = channelBindingServiceProperties;
this.messageBuilderFactory = messageBuilderFactory;
}
@Override
public void configureMessageChannel(MessageChannel messageChannel, String channelName) {
BindingProperties bindingProperties = channelBindingServiceProperties.getBindingProperties(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() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
@SuppressWarnings("unchecked")
Collection<Map<String, Object>> history =
(Collection<Map<String, Object>>) message.getHeaders().get(HISTORY_TRACKING_HEADER);
if (history == null) {
history = new ArrayList<>(1);
}
else {
history = new ArrayList<>(history);
}
Map<String, Object> map = new LinkedHashMap<String, Object>();
map.put("thread", Thread.currentThread().getName());
map.putAll(historyMap);
history.add(map);
Message<?> out = messageBuilderFactory
.fromMessage(message)
.setHeader(HISTORY_TRACKING_HEADER, history)
.build();
map.put("timestamp", out.getHeaders().getTimestamp());
return out;
}
});
}
}
}
}

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

@@ -49,18 +49,6 @@ 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";
public String getDestination() {
return this.destination;
}
@@ -93,22 +81,6 @@ public class BindingProperties {
this.binder = binder;
}
public Boolean isTrackHistory() {
return this.trackHistory;
}
public void setTrackHistory(Boolean trackHistory) {
this.trackHistory = trackHistory;
}
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);
@@ -123,10 +95,6 @@ public class BindingProperties {
sb.append("binder=" + this.binder);
sb.append(COMMA);
}
if (this.trackHistory != null) {
sb.append("trackHistory=" + this.trackHistory);
sb.append(COMMA);
}
sb.deleteCharAt(sb.lastIndexOf(COMMA));
return "BindingProperties{" + sb.toString() + "}";
}

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

@@ -35,7 +35,6 @@ import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binding.BindableChannelFactory;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor;
import org.springframework.cloud.stream.binding.ChannelBindingService;
import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer;
import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
@@ -44,9 +43,9 @@ import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.binding.MessageHistoryTrackerConfigurer;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
import org.springframework.cloud.stream.binding.SingleChannelBindable;
import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor;
import org.springframework.cloud.stream.converter.AbstractFromMessageConverter;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.context.annotation.Bean;
@@ -113,18 +112,11 @@ public class ChannelBindingServiceConfiguration {
return new DefaultBindableChannelFactory(compositeMessageChannelConfigurer);
}
@Bean
public MessageHistoryTrackerConfigurer messageHistoryTrackerConfigurer
(ChannelBindingServiceProperties channelBindingServiceProperties) {
return new MessageHistoryTrackerConfigurer(channelBindingServiceProperties, messageBuilderFactory);
}
@Bean
public CompositeMessageChannelConfigurer compositeMessageChannelConfigurer
(MessageConverterConfigurer messageConverterConfigurer, MessageHistoryTrackerConfigurer messageHistoryTrackerConfigurer) {
(MessageConverterConfigurer messageConverterConfigurer) {
List<MessageChannelConfigurer> configurerList = new ArrayList<>();
configurerList.add(messageConverterConfigurer);
configurerList.add((messageHistoryTrackerConfigurer));
return new CompositeMessageChannelConfigurer(configurerList);
}

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

@@ -1,130 +0,0 @@
/*
* 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());
}
}