From 95b0197dd7e48fda51dbe81a9e79c969c60ecfe1 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Thu, 17 Mar 2016 18:11:34 -0400 Subject: [PATCH] Removing message history support --- .../config/MessageChannelConfigurerTests.java | 29 +--- .../sink-channel-configurers.properties | 2 - .../MessageHistoryTrackerConfigurer.java | 99 ------------- .../stream/config/BindingProperties.java | 32 ----- .../ChannelBindingServiceConfiguration.java | 12 +- .../MessageHistoryTrackerConfigurerTests.java | 130 ------------------ 6 files changed, 3 insertions(+), 301 deletions(-) delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurerTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index 11553c9c1..8bccacb31 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java @@ -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 headerValue = ((Map) ((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") diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties index 2ea698490..26507d4ab 100644 --- a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties @@ -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 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 deleted file mode 100644 index 267881706..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java +++ /dev/null @@ -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 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() { - - @Override - public Message preSend(Message message, MessageChannel channel) { - @SuppressWarnings("unchecked") - Collection> history = - (Collection>) message.getHeaders().get(HISTORY_TRACKING_HEADER); - if (history == null) { - history = new ArrayList<>(1); - } - else { - history = new ArrayList<>(history); - } - Map map = new LinkedHashMap(); - 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; - } - }); - } - } - } -} 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 01057f4b6..0f3178c85 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 @@ -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() + "}"; } 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 bf5c24328..7f3cd22d0 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 @@ -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 configurerList = new ArrayList<>(); configurerList.add(messageConverterConfigurer); - configurerList.add((messageHistoryTrackerConfigurer)); return new CompositeMessageChannelConfigurer(configurerList); } 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 deleted file mode 100644 index b4b93a2cd..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurerTests.java +++ /dev/null @@ -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 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()); - } -}