diff --git a/pom.xml b/pom.xml index ea89a0a46..9c7586430 100644 --- a/pom.xml +++ b/pom.xml @@ -33,6 +33,7 @@ spring-cloud-stream-module-launcher spring-cloud-stream-test-support spring-cloud-stream-test-support-internal + spring-cloud-stream-integration-tests spring-cloud-stream-configuration-metadata spring-cloud-stream-docs diff --git a/spring-cloud-stream-integration-tests/pom.xml b/spring-cloud-stream-integration-tests/pom.xml new file mode 100644 index 000000000..5fcfe5164 --- /dev/null +++ b/spring-cloud-stream-integration-tests/pom.xml @@ -0,0 +1,42 @@ + + + 4.0.0 + + spring-cloud-stream-integration-tests + jar + spring-cloud-stream-integration-tests + Integration tests for Spring Cloud Stream + + + org.springframework.cloud + spring-cloud-stream-parent + 1.0.0.BUILD-SNAPSHOT + + + UTF-8 + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.cloud + spring-cloud-starter-stream-redis + 1.0.0.BUILD-SNAPSHOT + test + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + test + + + org.springframework.boot + spring-boot-starter-test + test + + + + 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 new file mode 100644 index 000000000..c5ead9894 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java @@ -0,0 +1,133 @@ +/* + * Copyright 2015 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.config; + +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; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.cloud.stream.annotation.Bindings; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.redis.config.RedisMessageChannelBinderConfiguration; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport; +import org.springframework.cloud.stream.tuple.Tuple; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.PropertySource; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration({MessageChannelConfigurerTests.TestSource.class, + MessageChannelConfigurerTests.TestSink.class}) +public class MessageChannelConfigurerTests { + + @Rule + public RedisTestSupport redisTestSupport = new RedisTestSupport(); + + @Autowired @Bindings(TestSink.class) + private Sink testSink; + + @Autowired + @Bindings(TestSource.class) + private Source testSource; + + @Test + public void testContentTypeConfigurer() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler messageHandler = new MessageHandler() { + @Override + public void handleMessage(Message message) throws MessagingException { + assertThat(message.getPayload(), instanceOf(Tuple.class)); + assertTrue(((Tuple) message.getPayload()).getFieldNames().get(0).equals("message")); + assertTrue(((Tuple) message.getPayload()).getValue(0).equals("Hi")); + latch.countDown(); + } + }; + testSink.input().subscribe(messageHandler); + testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").build()); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + testSink.input().unsubscribe(messageHandler); + } + + @Test + public void testHistoryTrackerConfigurer() throws Exception { + final CountDownLatch latch1 = new CountDownLatch(1); + MessageHandler messageHandler1 = 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")); + assertTrue(headerValue.get("producer.nextModuleCount").equals("1")); + assertTrue(headerValue.get("consumer.concurrency").equals("1")); + String outputBindingProps = (String) ((Map) ((List) message.getHeaders() + .get("SPRING_CLOUD_STREAM_HISTORY")).get(0)).get("output"); + ; + assertTrue(outputBindingProps.contains("destination=configure")); + assertTrue(outputBindingProps.contains("trackHistory=false")); + latch1.countDown(); + } + }; + testSink.input().subscribe(messageHandler1); + testSource.output().send(MessageBuilder.withPayload("{\"test\":\"value\"}").build()); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + testSink.input().unsubscribe(messageHandler1); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @Import(RedisMessageChannelBinderConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/config/source-channel-configurers.properties") + public static class TestSource { + + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + @Import(RedisMessageChannelBinderConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/config/sink-channel-configurers.properties") + public static class TestSink { + + } +} + diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/sink-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/sink-channel-configurers.properties new file mode 100644 index 000000000..67edfec91 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/sink-channel-configurers.properties @@ -0,0 +1,4 @@ +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.consumerProperties.concurrency=1 diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/source-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/source-channel-configurers.properties new file mode 100644 index 000000000..7a2b1b5bd --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/source-channel-configurers.properties @@ -0,0 +1,3 @@ +spring.cloud.stream.bindings.output.destination=configure1 +spring.cloud.stream.producerProperties.nextModuleCount=1 +spring.cloud.stream.bindings.output.contentType=application/json diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java new file mode 100644 index 000000000..b51718f5f --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java @@ -0,0 +1,41 @@ +/* + * Copyright 2015 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.List; + +import org.springframework.messaging.MessageChannel; + +/** + * {@link MessageChannelConfigurer} that composes all the message channel configurers. + * + * @author Ilayaperumal Gopinathan + */ +public class CompositeMessageChannelConfigurer implements MessageChannelConfigurer { + + private final List messageChannelConfigurers; + + public CompositeMessageChannelConfigurer(List messageChannelConfigurers) { + this.messageChannelConfigurers = messageChannelConfigurers; + } + + @Override + public void configureMessageChannel(MessageChannel messageChannel, String channelName) { + for (MessageChannelConfigurer messageChannelConfigurer : messageChannelConfigurers) { + messageChannelConfigurer.configureMessageChannel(messageChannel, channelName); + } + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultBindableChannelFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultBindableChannelFactory.java index c14bee203..109fd00ad 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultBindableChannelFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultBindableChannelFactory.java @@ -29,23 +29,23 @@ import org.springframework.messaging.SubscribableChannel; */ public class DefaultBindableChannelFactory implements BindableChannelFactory { - private final MessageConverterConfigurer messageConverterConfigurer; + private final MessageChannelConfigurer messageChannelConfigurer; - public DefaultBindableChannelFactory(MessageConverterConfigurer messageConverterConfigurer) { - this.messageConverterConfigurer = messageConverterConfigurer; + public DefaultBindableChannelFactory(MessageChannelConfigurer messageChannelConfigurer) { + this.messageChannelConfigurer = messageChannelConfigurer; } @Override public PollableChannel createPollableChannel(String name) { PollableChannel pollableChannel = new QueueChannel(); - messageConverterConfigurer.configureMessageConverters(pollableChannel, name); + messageChannelConfigurer.configureMessageChannel(pollableChannel, name); return pollableChannel; } @Override public SubscribableChannel createSubscribableChannel(String name) { SubscribableChannel subscribableChannel = new DirectChannel(); - messageConverterConfigurer.configureMessageConverters(subscribableChannel, name); + messageChannelConfigurer.configureMessageChannel(subscribableChannel, name); return subscribableChannel; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelConfigurer.java new file mode 100644 index 000000000..11123df02 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelConfigurer.java @@ -0,0 +1,34 @@ +/* + * Copyright 2015 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 org.springframework.messaging.MessageChannel; + +/** + * Interface to be implemented by the classes that configure the {@link Bindable} message channels. + * + * @author Ilayaperumal Gopinathan + */ +public interface MessageChannelConfigurer { + + /** + * Configure the given message channel. + * + * @param messageChannel the message channel + * @param channelName name of the message channel + */ + void configureMessageChannel(MessageChannel messageChannel, String channelName); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index deb5ff544..f66dd5a20 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -48,7 +48,7 @@ import org.springframework.util.StringUtils; * * @author Ilayaperumal Gopinathan */ -public class MessageConverterConfigurer implements BeanFactoryAware, InitializingBean { +public class MessageConverterConfigurer implements MessageChannelConfigurer, BeanFactoryAware, InitializingBean { private ConfigurableListableBeanFactory beanFactory; @@ -87,7 +87,8 @@ public class MessageConverterConfigurer implements BeanFactoryAware, Initializin * @param channel message channel to set the data-type and message converters * @param channelName the channel name */ - void configureMessageConverters(MessageChannel channel, String channelName) { + @Override + public void configureMessageChannel(MessageChannel channel, String channelName) { Assert.isAssignable(AbstractMessageChannel.class, channel.getClass()); AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel; BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindings().get(channelName); 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 new file mode 100644 index 000000000..50efbd070 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java @@ -0,0 +1,82 @@ +/* + * Copyright 2015 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.LinkedHashMap; +import java.util.Map; + +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; +import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.ChannelInterceptorAdapter; + +/** + * 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; + + @Autowired + MessageBuilderFactory messageBuilderFactory; + + public MessageHistoryTrackerConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties) { + this.channelBindingServiceProperties = channelBindingServiceProperties; + } + + @Override + public void configureMessageChannel(MessageChannel messageChannel, String channelName) { + BindingProperties bindingProperties = channelBindingServiceProperties.getBindings().get(channelName); + if (bindingProperties != null && Boolean.TRUE.equals(bindingProperties.getTrackHistory())) { + 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()); + history.add(channelBindingServiceProperties.asMapProperties()); + 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 40c3e9e7b..c1d257eb0 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 @@ -31,6 +31,8 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include; @JsonInclude(value = Include.NON_DEFAULT) public class BindingProperties { + private static final String COMMA = ","; + private String destination; private boolean partitioned = false; @@ -51,6 +53,8 @@ public class BindingProperties { private String binder; + private boolean trackHistory; + public String getDestination() { return this.destination; } @@ -123,6 +127,7 @@ public class BindingProperties { this.contentType = contentType; } + public String getBinder() { return binder; } @@ -130,4 +135,46 @@ public class BindingProperties { public void setBinder(String binder) { this.binder = binder; } + + public Boolean getTrackHistory() { + return this.trackHistory; + } + + public void setTrackHistory(boolean trackHistory) { + this.trackHistory = trackHistory; + } + + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append("destination=" + destination); + sb.append(COMMA); + sb.append("group=" + group); + sb.append(COMMA); + sb.append("contentType="+ contentType); + sb.append(COMMA); + sb.append("binder="+ binder); + sb.append(COMMA); + sb.append("trackHistory=" + trackHistory); + sb.append(COMMA); + sb.append("partitioned=" + partitioned); + sb.append(COMMA); + if (this.partitionKeyExpression != null && !this.partitionKeyExpression.isEmpty()) { + sb.append("partitionKeyExpression=" + partitionKeyExpression); + sb.append(COMMA); + } + if (this.partitionKeyExtractorClass != null && !this.partitionKeyExtractorClass.isEmpty()) { + sb.append("partitionKeyExtractorClass=" + partitionKeyExtractorClass); + sb.append(COMMA); + } + if (this.partitionSelectorClass != null && !this.partitionSelectorClass.isEmpty()) { + sb.append("partitionSelectorClass=" + partitionSelectorClass); + sb.append(COMMA); + } + if (this.partitionSelectorClass != null && !this.partitionSelectorClass.isEmpty()) { + sb.append("partitionSelectorExpression=" + partitionSelectorExpression); + } + sb.deleteCharAt(sb.lastIndexOf(COMMA)); + return "BinderProperties{" + 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 c150d73e7..85f664ab2 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 @@ -16,7 +16,9 @@ package org.springframework.cloud.stream.config; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Properties; @@ -28,14 +30,17 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.context.properties.ConfigurationPropertiesBinding; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binding.BindableChannelFactory; +import org.springframework.cloud.stream.binding.MessageChannelConfigurer; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor; import org.springframework.cloud.stream.binding.ChannelBindingService; -import org.springframework.cloud.stream.binding.BindableChannelFactory; +import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer; import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener; import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory; import org.springframework.cloud.stream.binding.InputBindingLifecycle; 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.tuple.spel.TuplePropertyAccessor; import org.springframework.context.annotation.Bean; @@ -74,13 +79,29 @@ public class ChannelBindingServiceConfiguration { } @Bean - public MessageConverterConfigurer messageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties) { + public MessageConverterConfigurer messageConverterConfigurer + (ChannelBindingServiceProperties channelBindingServiceProperties) { return new MessageConverterConfigurer(channelBindingServiceProperties); } @Bean public BindableChannelFactory channelFactory(ChannelBindingServiceProperties channelBindingServiceProperties) { - return new DefaultBindableChannelFactory(messageConverterConfigurer(channelBindingServiceProperties)); + return new DefaultBindableChannelFactory(compositeMessageChannelConfigurer(channelBindingServiceProperties)); + } + + @Bean + public MessageHistoryTrackerConfigurer messageHistoryTrackerConfigurer + (ChannelBindingServiceProperties channelBindingServiceProperties) { + return new MessageHistoryTrackerConfigurer(channelBindingServiceProperties); + } + + @Bean + public CompositeMessageChannelConfigurer compositeMessageChannelConfigurer + (ChannelBindingServiceProperties channelBindingServiceProperties) { + List configurerList = new ArrayList<>(); + configurerList.add(messageConverterConfigurer(channelBindingServiceProperties)); + configurerList.add((messageHistoryTrackerConfigurer(channelBindingServiceProperties))); + return new CompositeMessageChannelConfigurer(configurerList); } @Bean @@ -109,7 +130,7 @@ public class ChannelBindingServiceConfiguration { @Bean @ConfigurationPropertiesBinding - public Converter bindingPropertiesConverter() { + public Converter bindingPropertiesConverter() { return new BindingPropertiesConverter(); } 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 2a2a9dfd4..bd723336e 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 @@ -21,13 +21,13 @@ import java.util.Map; import java.util.Properties; import java.util.TreeMap; -import com.fasterxml.jackson.annotation.JsonInclude; -import com.fasterxml.jackson.annotation.JsonInclude.Include; - import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.util.StringUtils; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonInclude.Include; + /** * @author Dave Syer * @author Marius Bogoevici @@ -127,8 +127,8 @@ public class ChannelBindingServiceProperties { public boolean isPartitionedProducer(String channelName) { BindingProperties bindingProperties = bindings.get(channelName); return bindingProperties != null && - (StringUtils.hasText(bindingProperties.getPartitionKeyExpression()) - || StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass())); + (StringUtils.hasText(bindingProperties.getPartitionKeyExpression()) + || StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass())); } @@ -200,4 +200,26 @@ public class ChannelBindingServiceProperties { return bindings.get(channelName).getBinder(); } + /** + * Return configuration properties as Map. + * @return map of channel binding configuration properties. + */ + public Map asMapProperties() { + Map properties = new HashMap<>(); + properties.put("instanceIndex", String.valueOf(getInstanceIndex())); + properties.put("instanceCount", String.valueOf(getInstanceCount())); + Properties consumerProperties = getConsumerProperties(); + for (String name : consumerProperties.stringPropertyNames()) { + properties.put("consumer." + name, consumerProperties.getProperty(name)); + } + Properties producerProperties = getProducerProperties(); + for (String name : producerProperties.stringPropertyNames()) { + properties.put("producer." + name, producerProperties.getProperty(name)); + } + for (Map.Entry entry : getBindings().entrySet()) { + properties.put(entry.getKey(), entry.getValue().toString()); + } + return properties; + } + }