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

GH-196 Add Message History Tracking

Browse Source 
- Add `MessageChannelConfigurer` that any class that configures the message channel would implement
   - Have `message-converter` and `message-history-tracking` configurer implement this interface
  - Autowire all available configurers and use them to configure the message channel
  - For the message history tracking
    - add `channel-binding` properties as part of message header along with `timestamp`

 - Introduce CompositeMessageChannelConfigurer

Remove field injection of messageChannelConfigurer

Add tests for message track history configurer

Add content-type configurer test

Address review comments

 - Add `toString` to BindingProperties
 - Add test case to depict producer/consumer properties

Remove processor channels from tests
This commit is contained in:
Ilayaperumal Gopinathan
2015-12-03 17:32:25 +05:30
committed by Gary Russell
parent 001ad52f4a
commit 100a68c1bf
13 changed files with 447 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pom.xml
View File

@@ -33,6 +33,7 @@
<module>spring-cloud-stream-module-launcher</module>
<module>spring-cloud-stream-test-support</module>
<module>spring-cloud-stream-test-support-internal</module>
<module>spring-cloud-stream-integration-tests</module>
<module>spring-cloud-stream-configuration-metadata</module>
<module>spring-cloud-stream-docs</module>
</modules>

42
spring-cloud-stream-integration-tests/pom.xml Normal file
View File

@@ -0,0 +1,42 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-integration-tests</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-integration-tests</name>
<description>Integration tests for Spring Cloud Stream</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-redis</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

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

@@ -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<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"));
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 {
}
}

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

@@ -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

3
spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/source-channel-configurers.properties Normal file
View File

@@ -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

41
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java Normal file
View File

@@ -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<MessageChannelConfigurer> messageChannelConfigurers;
public CompositeMessageChannelConfigurer(List<MessageChannelConfigurer> messageChannelConfigurers) {
this.messageChannelConfigurers = messageChannelConfigurers;
}
@Override
public void configureMessageChannel(MessageChannel messageChannel, String channelName) {
for (MessageChannelConfigurer messageChannelConfigurer : messageChannelConfigurers) {
messageChannelConfigurer.configureMessageChannel(messageChannel, channelName);
}
}
}

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultBindableChannelFactory.java
View File

@@ -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;
}
}

34
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelConfigurer.java Normal file
View File

@@ -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);
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -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);

82
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageHistoryTrackerConfigurer.java Normal file
View File

@@ -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<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());
history.add(channelBindingServiceProperties.asMapProperties());
Message<?> out = messageBuilderFactory
.fromMessage(message)
.setHeader(HISTORY_TRACKING_HEADER, history)
.build();
map.put("timestamp", out.getHeaders().getTimestamp());
return out;
}
});
}
}
}
}

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

@@ -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() + "}";
}
}

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

@@ -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<MessageChannelConfigurer> 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<String,BindingProperties> bindingPropertiesConverter() {
public Converter<String, BindingProperties> bindingPropertiesConverter() {
return new BindingPropertiesConverter();
}

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

@@ -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<String, Object> asMapProperties() {
Map<String, Object> 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<String, BindingProperties> entry : getBindings().entrySet()) {
properties.put(entry.getKey(), entry.getValue().toString());
}
return properties;
}
}