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

Handle binder config in BindingProperties

Browse Source 
- Per-binding properties defined in `BindingProperties` need to hold the binder specific configurations needed when binding producer/consumers.
Previously, these were added in `ChannelBindingServiceProperties` as producer/consumer properties. Since these configuration properties are for
per-binding based on the input/output (consumer/producer) channel being bound it is better to move these properties into `BindingProperties`.
 - Add some of the supported producer/consumer properties into BindingProperties

- This fixes #256

Rename CommonBinderProperties -> BinderPropertyKeys

Rename class AbstractBinderPropertiesAccessor.java -> AbstractBindingPropertiesAccessor.java
This commit is contained in:
Ilayaperumal Gopinathan
2015-12-22 22:41:28 +05:30
committed by Mark Fisher
parent 373dd12fbb
commit fa89e5bcb6
22 changed files with 408 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -37,10 +37,10 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.cloud.stream.binder.AbstractBinderPropertiesAccessor;
import org.springframework.cloud.stream.binder.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderException;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
@@ -191,7 +191,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private static final String POINT_TO_POINT_SEMANTICS_CONSUMER_GROUP = "springXD";
private static final Set<Object> KAFKA_CONSUMER_PROPERTIES = new SetBuilder()
.add(BinderProperties.MIN_PARTITION_COUNT)
.add(BinderPropertyKeys.MIN_PARTITION_COUNT)
.build();
/**
@@ -200,14 +200,14 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private static final Set<Object> SUPPORTED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(CONSUMER_STANDARD_PROPERTIES)
.addAll(KAFKA_CONSUMER_PROPERTIES)
.add(BinderProperties.PARTITION_INDEX) // Not actually used
.add(BinderProperties.COUNT) // Not actually used
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.PARTITION_INDEX) // Not actually used
.add(BinderPropertyKeys.COUNT) // Not actually used
.add(BinderPropertyKeys.CONCURRENCY)
.add(FETCH_SIZE)
.build();
private static final Set<Object> KAFKA_PRODUCER_PROPERTIES = new SetBuilder()
.add(BinderProperties.MIN_PARTITION_COUNT)
.add(BinderPropertyKeys.MIN_PARTITION_COUNT)
.build();
/**
@@ -228,7 +228,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private static final Set<Object> SUPPORTED_PRODUCER_PROPERTIES = new SetBuilder()
.addAll(PRODUCER_PARTITIONING_PROPERTIES)
.addAll(PRODUCER_STANDARD_PROPERTIES)
.add(BinderProperties.DIRECT_BINDING_ALLOWED)
.add(BinderPropertyKeys.DIRECT_BINDING_ALLOWED)
.addAll(KAFKA_PRODUCER_PROPERTIES)
.addAll(PRODUCER_BATCHING_BASIC_PROPERTIES)
.addAll(PRODUCER_COMPRESSION_PROPERTIES)
@@ -787,7 +787,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
}
}
private class KafkaPropertiesAccessor extends AbstractBinderPropertiesAccessor {
private class KafkaPropertiesAccessor extends AbstractBindingPropertiesAccessor {
public KafkaPropertiesAccessor(Properties properties) {
super(properties);
@@ -798,7 +798,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
if (nextModuleCount == 0) {
throw new IllegalArgumentException("Module count cannot be zero");
}
int nextModuleConcurrency = getProperty(NEXT_MODULE_CONCURRENCY, defaultConcurrency);
int nextModuleConcurrency = getProperty(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, defaultConcurrency);
int minKafkaPartitions = getMinPartitionCount(defaultMinPartitionCount);
return Math.max(minKafkaPartitions, nextModuleCount * nextModuleConcurrency);
}
@@ -826,7 +826,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
}
public int getMinPartitionCount(int defaultPartitionCount) {
return getProperty(MIN_PARTITION_COUNT, defaultPartitionCount);
return getProperty(BinderPropertyKeys.MIN_PARTITION_COUNT, defaultPartitionCount);
}
}

38
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -35,7 +35,7 @@ import org.junit.Ignore;
import org.junit.Test;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
@@ -162,9 +162,9 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Properties producerProperties = new Properties();
producerProperties.put(BinderProperties.MIN_PARTITION_COUNT, "10");
producerProperties.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "10");
Properties consumerProperties = new Properties();
consumerProperties.put(BinderProperties.MIN_PARTITION_COUNT, "10");
consumerProperties.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "10");
long uniqueBindingId = System.currentTimeMillis();
binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
binder.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, consumerProperties);
@@ -193,11 +193,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Properties producerProps = new Properties();
producerProps.put(BinderProperties.MIN_PARTITION_COUNT, "5");
producerProps.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "6");
producerProps.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "5");
producerProps.put(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, "6");
Properties consumerProps = new Properties();
consumerProps.put(BinderProperties.MIN_PARTITION_COUNT, "5");
consumerProps.put(BinderProperties.CONCURRENCY, "6");
consumerProps.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "5");
consumerProps.put(BinderPropertyKeys.CONCURRENCY, "6");
long uniqueBindingId = System.currentTimeMillis();
binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProps);
binder.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, consumerProps);
@@ -225,11 +225,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Properties producerProps = new Properties();
producerProps.put(BinderProperties.MIN_PARTITION_COUNT, "6");
producerProps.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "5");
producerProps.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "6");
producerProps.put(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, "5");
Properties consumerProps = new Properties();
consumerProps.put(BinderProperties.MIN_PARTITION_COUNT, "6");
consumerProps.put(BinderProperties.CONCURRENCY, "5");
consumerProps.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "6");
consumerProps.put(BinderPropertyKeys.CONCURRENCY, "5");
long uniqueBindingId = System.currentTimeMillis();
binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProps);
binder.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, consumerProps);
@@ -257,11 +257,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Properties producerProperties = new Properties();
producerProperties.put(BinderProperties.MIN_PARTITION_COUNT, "3");
producerProperties.put(BinderProperties.NEXT_MODULE_COUNT, "5");
producerProperties.put(BinderProperties.PARTITION_KEY_EXPRESSION, "payload");
producerProperties.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "3");
producerProperties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "5");
producerProperties.put(BinderPropertyKeys.PARTITION_KEY_EXPRESSION, "payload");
Properties consumerProperties = new Properties();
consumerProperties.put(BinderProperties.MIN_PARTITION_COUNT, "3");
consumerProperties.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "3");
long uniqueBindingId = System.currentTimeMillis();
binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
binder.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, consumerProperties);
@@ -289,11 +289,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Properties producerProperties = new Properties();
producerProperties.put(BinderProperties.MIN_PARTITION_COUNT, "5");
producerProperties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
producerProperties.put(BinderProperties.PARTITION_KEY_EXPRESSION, "payload");
producerProperties.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "5");
producerProperties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "3");
producerProperties.put(BinderPropertyKeys.PARTITION_KEY_EXPRESSION, "payload");
Properties consumerProperties = new Properties();
consumerProperties.put(BinderProperties.MIN_PARTITION_COUNT, "5");
consumerProperties.put(BinderPropertyKeys.MIN_PARTITION_COUNT, "5");
long uniqueBindingId = System.currentTimeMillis();
binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
binder.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, consumerProperties);

10
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java
View File

@@ -34,7 +34,7 @@ import org.junit.Test;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.TestUtils;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -69,8 +69,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
Properties properties = new Properties();
properties.put("partitionKeyExtractorClass", "org.springframework.cloud.stream.binder.kafka.RawKafkaPartitionTestSupport");
properties.put("partitionSelectorClass", "org.springframework.cloud.stream.binder.kafka.RawKafkaPartitionTestSupport");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
properties.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "2");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "3");
properties.put(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, "2");
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
@@ -123,8 +123,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
Properties properties = new Properties();
properties.put("partitionKeyExpression", "payload[0]");
properties.put("partitionSelectorExpression", "hashCode()");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
properties.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "2");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "3");
properties.put(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, "2");
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");

24
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -59,8 +59,8 @@ import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.cloud.stream.binder.AbstractBinderPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.BinderUtils;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
@@ -128,7 +128,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
private static final Set<Object> RABBIT_CONSUMER_PROPERTIES = new HashSet<Object>(Arrays.asList(new String[] {
BinderProperties.MAX_CONCURRENCY,
BinderPropertyKeys.MAX_CONCURRENCY,
RabbitPropertiesAccessor.ACK_MODE,
RabbitPropertiesAccessor.PREFETCH,
RabbitPropertiesAccessor.PREFIX,
@@ -151,7 +151,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
private static final Set<Object> SUPPORTED_PUBSUB_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(SUPPORTED_BASIC_CONSUMER_PROPERTIES)
.add(BinderProperties.DURABLE)
.add(BinderPropertyKeys.DURABLE)
.build();
/**
@@ -159,7 +159,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
*/
private static final Set<Object> SUPPORTED_NAMED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(SUPPORTED_BASIC_CONSUMER_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
/**
@@ -167,8 +167,8 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
*/
private static final Set<Object> SUPPORTED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(SUPPORTED_BASIC_CONSUMER_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderProperties.PARTITION_INDEX)
.add(BinderPropertyKeys.CONCURRENCY)
.add(BinderPropertyKeys.PARTITION_INDEX)
.build();
/**
@@ -177,7 +177,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
private static final Set<Object> SUPPORTED_REPLYING_CONSUMER_PROPERTIES = new SetBuilder()
// request
.addAll(SUPPORTED_BASIC_CONSUMER_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
// reply
.add(RabbitPropertiesAccessor.REPLY_HEADER_PATTERNS)
.add(RabbitPropertiesAccessor.DELIVERY_MODE)
@@ -191,7 +191,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
.add(RabbitPropertiesAccessor.DELIVERY_MODE)
.add(RabbitPropertiesAccessor.PREFIX)
.add(RabbitPropertiesAccessor.REQUEST_HEADER_PATTERNS)
.add(BinderProperties.COMPRESS)
.add(BinderPropertyKeys.COMPRESS)
.build();
private static final Set<Object> SUPPORTED_PUBSUB_PRODUCER_PROPERTIES = new SetBuilder()
@@ -212,7 +212,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
private static final Set<Object> SUPPORTED_PRODUCER_PROPERTIES = new SetBuilder()
.addAll(PRODUCER_PARTITIONING_PROPERTIES)
.addAll(SUPPORTED_BASIC_PRODUCER_PROPERTIES)
.add(BinderProperties.DIRECT_BINDING_ALLOWED)
.add(BinderPropertyKeys.DIRECT_BINDING_ALLOWED)
.addAll(PRODUCER_BATCHING_BASIC_PROPERTIES)
.addAll(PRODUCER_BATCHING_ADVANCED_PROPERTIES)
.build();
@@ -225,7 +225,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
.addAll(SUPPORTED_BASIC_PRODUCER_PROPERTIES)
// reply
.addAll(SUPPORTED_BASIC_CONSUMER_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
.add(RabbitPropertiesAccessor.REPLY_HEADER_PATTERNS)
.build();
@@ -958,7 +958,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
* Property accessor for the RabbitBinder. Refer to the Spring-AMQP documentation for information on the
* specific properties.
*/
private static class RabbitPropertiesAccessor extends AbstractBinderPropertiesAccessor {
private static class RabbitPropertiesAccessor extends AbstractBindingPropertiesAccessor {
/**
* The acknowledge mode (i.e. NONE, MANUAL, AUTO).

8
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -60,7 +60,7 @@ import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.Spy;
@@ -230,7 +230,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "1");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
binder.bindProducer("props.0", new DirectChannel(), properties);
assertEquals(1, bindings.size());
@@ -318,7 +318,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "1");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
properties.put("partitionIndex", "0");
try {
binder.bindRequestor("dummy", null, null, properties);
@@ -387,7 +387,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "1");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
properties.put("partitionIndex", "0");
try {
binder.bindReplier("dummy", null, null, properties);

16
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java
View File

@@ -23,9 +23,9 @@ import java.util.Properties;
import java.util.Set;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.cloud.stream.binder.AbstractBinderPropertiesAccessor;
import org.springframework.cloud.stream.binder.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
@@ -83,7 +83,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
private static final Set<Object> SUPPORTED_NAMED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(CONSUMER_STANDARD_PROPERTIES)
.addAll(CONSUMER_RETRY_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
/**
@@ -91,7 +91,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
*/
private static final Set<Object> SUPPORTED_CONSUMER_PROPERTIES = new SetBuilder()
.addAll(SUPPORTED_NAMED_CONSUMER_PROPERTIES)
.add(BinderProperties.PARTITION_INDEX)
.add(BinderPropertyKeys.PARTITION_INDEX)
.build();
/**
@@ -101,7 +101,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
// request
.addAll(CONSUMER_STANDARD_PROPERTIES)
.addAll(CONSUMER_RETRY_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
/**
@@ -120,7 +120,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
private static final Set<Object> SUPPORTED_PRODUCER_PROPERTIES = new SetBuilder()
.addAll(PRODUCER_PARTITIONING_PROPERTIES)
.addAll(PRODUCER_STANDARD_PROPERTIES)
.add(BinderProperties.DIRECT_BINDING_ALLOWED)
.add(BinderPropertyKeys.DIRECT_BINDING_ALLOWED)
.build();
/**
@@ -129,7 +129,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
private static final Set<Object> SUPPORTED_REQUESTING_PRODUCER_PROPERTIES = new SetBuilder()
// reply
.addAll(CONSUMER_RETRY_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
private final RedisConnectionFactory connectionFactory;
@@ -466,7 +466,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
}
private static class RedisPropertiesAccessor extends AbstractBinderPropertiesAccessor {
private static class RedisPropertiesAccessor extends AbstractBindingPropertiesAccessor {
public RedisPropertiesAccessor(Properties properties) {
super(properties);

6
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java
View File

@@ -39,7 +39,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
@@ -171,7 +171,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "1");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
binder.bindProducer("props.0", new DirectChannel(), properties);
assertEquals(1, bindings.size());
@@ -287,7 +287,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
properties.put("partitionKeyExtractorClass", "foo");
properties.put("partitionSelectorExpression", "0");
properties.put("partitionSelectorClass", "foo");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "1");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1");
properties.put("partitionIndex", "0");
try {
binder.bindReplier("dummy", new DirectChannel(), new DirectChannel(), properties);

2
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -247,7 +247,7 @@ public abstract class AbstractBinderTests {
@Test
public void testBadDynamic() throws Exception {
Properties properties = new Properties();
properties.setProperty(BinderProperties.PARTITION_KEY_EXPRESSION, "'foo'");
properties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXPRESSION, "'foo'");
Binder<MessageChannel> binder = getBinder();
try {
binder.bindDynamicProducer("queue:foo", properties);

2
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/BrokerBinderTests.java
View File

@@ -46,7 +46,7 @@ public abstract class BrokerBinderTests extends
public void testDirectBinding() throws Exception {
Binder binder = getBinder();
Properties properties = new Properties();
properties.setProperty(BinderProperties.DIRECT_BINDING_ALLOWED, "true");
properties.setProperty(BinderPropertyKeys.DIRECT_BINDING_ALLOWED, "true");
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("direct.input");

8
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java
View File

@@ -86,8 +86,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
Properties properties = new Properties();
properties.put("partitionKeyExpression", "payload");
properties.put("partitionSelectorExpression", "hashCode()");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
properties.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "2");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "3");
properties.put(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, "2");
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
@@ -187,8 +187,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests {
Properties properties = new Properties();
properties.put("partitionKeyExtractorClass", "org.springframework.cloud.stream.binder.PartitionTestSupport");
properties.put("partitionSelectorClass", "org.springframework.cloud.stream.binder.PartitionTestSupport");
properties.put(BinderProperties.NEXT_MODULE_COUNT, "3");
properties.put(BinderProperties.NEXT_MODULE_CONCURRENCY, "2");
properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "3");
properties.put(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY, "2");
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");

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

@@ -96,15 +96,15 @@ public class MessageChannelConfigurerTests {
String inputBindingProps = headerValue.get("input");
assertTrue(inputBindingProps.contains("destination=configure"));
assertTrue(inputBindingProps.contains("trackHistory=true"));
assertTrue(inputBindingProps.contains("concurrency=1"));
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"));
assertTrue(!outputBindingProps.contains("trackHistory"));
assertTrue(outputBindingProps.contains("nextModuleCount=1"));
latch1.countDown();
}
};

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

@@ -1,4 +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
spring.cloud.stream.bindings.input.concurrency=1

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

@@ -1,3 +1,3 @@
spring.cloud.stream.bindings.output.destination=configure1
spring.cloud.stream.producerProperties.nextModuleCount=1
spring.cloud.stream.bindings.output.nextModuleCount=1
spring.cloud.stream.bindings.output.contentType=application/json

48
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderPropertiesAccessor.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java
View File

@@ -24,18 +24,18 @@ import org.springframework.util.StringUtils;
/**
* Base class for binder-specific property accessors; common properties
* Base class for binding-specific property accessors; common properties
* are defined here.
*
* @author Gary Russell
*/
public abstract class AbstractBinderPropertiesAccessor implements BinderProperties {
public abstract class AbstractBindingPropertiesAccessor {
private static final SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
private final Properties properties;
public AbstractBinderPropertiesAccessor(Properties properties) {
public AbstractBindingPropertiesAccessor(Properties properties) {
if (properties == null) {
this.properties = new Properties();
}
@@ -147,7 +147,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The property or default value.
*/
public int getConcurrency(int defaultValue) {
return getProperty(CONCURRENCY, defaultValue);
return getProperty(BinderPropertyKeys.CONCURRENCY, defaultValue);
}
/**
@@ -157,7 +157,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The property or default value.
*/
public int getMaxConcurrency(int defaultValue) {
return getProperty(MAX_CONCURRENCY, defaultValue);
return getProperty(BinderPropertyKeys.MAX_CONCURRENCY, defaultValue);
}
// Retry properties
@@ -170,7 +170,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The property or default value.
*/
public int getMaxAttempts(int defaultValue) {
return getProperty(MAX_ATTEMPTS, defaultValue);
return getProperty(BinderPropertyKeys.MAX_ATTEMPTS, defaultValue);
}
/**
@@ -181,7 +181,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The property or default value.
*/
public long getBackOffInitialInterval(long defaultValue) {
return getProperty(BACK_OFF_INITIAL_INTERVAL, defaultValue);
return getProperty(BinderPropertyKeys.BACK_OFF_INITIAL_INTERVAL, defaultValue);
}
/**
@@ -192,7 +192,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The property or default value.
*/
public double getBackOffMultiplier(double defaultValue) {
return getProperty(BACK_OFF_MULTIPLIER, defaultValue);
return getProperty(BinderPropertyKeys.BACK_OFF_MULTIPLIER, defaultValue);
}
/**
@@ -203,7 +203,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The property or default value.
*/
public long getBackOffMaxInterval(long defaultValue) {
return getProperty(BACK_OFF_MAX_INTERVAL, defaultValue);
return getProperty(BinderPropertyKeys.BACK_OFF_MAX_INTERVAL, defaultValue);
}
// Partitioning
@@ -213,7 +213,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The class name,
*/
public String getPartitionKeyExtractorClass() {
return getProperty(PARTITION_KEY_EXTRACTOR_CLASS);
return getProperty(BinderPropertyKeys.PARTITION_KEY_EXTRACTOR_CLASS);
}
/**
@@ -222,7 +222,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The key.
*/
public Expression getPartitionKeyExpression() {
String partionKeyExpression = getProperty(PARTITION_KEY_EXPRESSION);
String partionKeyExpression = getProperty(BinderPropertyKeys.PARTITION_KEY_EXPRESSION);
Expression expression = null;
if (partionKeyExpression != null) {
expression = spelExpressionParser.parseExpression(partionKeyExpression);
@@ -235,7 +235,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The class name,
*/
public String getPartitionSelectorClass() {
return getProperty(PARTITION_SELECTOR_CLASS);
return getProperty(BinderPropertyKeys.PARTITION_SELECTOR_CLASS);
}
/**
@@ -246,7 +246,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The expression.
*/
public Expression getPartitionSelectorExpression() {
String partionSelectorExpression = getProperty(PARTITION_SELECTOR_EXPRESSION);
String partionSelectorExpression = getProperty(BinderPropertyKeys.PARTITION_SELECTOR_EXPRESSION);
Expression expression = null;
if (partionSelectorExpression != null) {
expression = spelExpressionParser.parseExpression(partionSelectorExpression);
@@ -260,7 +260,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the sequence number.
*/
public int getSequence() {
return getProperty(SEQUENCE, 1);
return getProperty(BinderPropertyKeys.SEQUENCE, 1);
}
/**
@@ -269,7 +269,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the module count.
*/
public int getCount() {
return getProperty(COUNT, 1);
return getProperty(BinderPropertyKeys.COUNT, 1);
}
/**
@@ -277,7 +277,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the next module count
*/
public int getNextModuleCount() {
return getProperty(NEXT_MODULE_COUNT, 1);
return getProperty(BinderPropertyKeys.NEXT_MODULE_COUNT, 1);
}
/**
@@ -285,7 +285,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return The partition index.
*/
public int getPartitionIndex() {
return getProperty(PARTITION_INDEX, -1);
return getProperty(BinderPropertyKeys.PARTITION_INDEX, -1);
}
// Direct Binding
@@ -294,7 +294,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* If true, the binder can attempt a direct binding.
*/
public boolean isDirectBindingAllowed() {
return getProperty(DIRECT_BINDING_ALLOWED, false);
return getProperty(BinderPropertyKeys.DIRECT_BINDING_ALLOWED, false);
}
// Batching
@@ -305,7 +305,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the property or default value.
*/
public boolean isBatchingEnabled(boolean defaultValue) {
return getProperty(BATCHING_ENABLED, defaultValue);
return getProperty(BinderPropertyKeys.BATCHING_ENABLED, defaultValue);
}
/**
@@ -314,7 +314,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the property or default value.
*/
public int getBatchSize(int defaultValue) {
return getProperty(BATCH_SIZE, defaultValue);
return getProperty(BinderPropertyKeys.BATCH_SIZE, defaultValue);
}
/**
@@ -323,7 +323,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the property or default value.
*/
public int geteBatchBufferLimit(int defaultValue) {
return getProperty(BATCH_BUFFER_LIMIT, defaultValue);
return getProperty(BinderPropertyKeys.BATCH_BUFFER_LIMIT, defaultValue);
}
/**
@@ -332,7 +332,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the property or default value.
*/
public long getBatchTimeout(long defaultValue) {
return getProperty(BATCH_TIMEOUT, defaultValue);
return getProperty(BinderPropertyKeys.BATCH_TIMEOUT, defaultValue);
}
/**
@@ -341,7 +341,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the property or default value.
*/
public boolean isCompress(boolean defaultValue) {
return getProperty(COMPRESS, defaultValue);
return getProperty(BinderPropertyKeys.COMPRESS, defaultValue);
}
/**
@@ -350,7 +350,7 @@ public abstract class AbstractBinderPropertiesAccessor implements BinderProperti
* @return the property or default value.
*/
public boolean isDurable(boolean defaultValue) {
return getProperty(DURABLE, defaultValue);
return getProperty(BinderPropertyKeys.DURABLE, defaultValue);
}
// Utility methods

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderProperties.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java
View File

@@ -21,8 +21,9 @@ package org.springframework.cloud.stream.binder;
* Common binder properties.
*
* @author Gary Russell
* @author Ilayaperumal Gopinathan
*/
public interface BinderProperties {
public abstract class BinderPropertyKeys {
/**
* The retry back off initial interval.

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
View File

@@ -47,10 +47,10 @@ public class Binding implements Lifecycle {
private final String type;
private final AbstractBinderPropertiesAccessor properties;
private final AbstractBindingPropertiesAccessor properties;
private Binding(String name, MessageChannel channel, AbstractEndpoint endpoint, String type,
AbstractBinderPropertiesAccessor properties) {
AbstractBindingPropertiesAccessor properties) {
Assert.notNull(channel, "channel must not be null");
Assert.notNull(endpoint, "endpoint must not be null");
this.name = name;
@@ -61,17 +61,17 @@ public class Binding implements Lifecycle {
}
public static Binding forConsumer(String name, AbstractEndpoint adapterFromBinder, MessageChannel moduleInputChannel,
AbstractBinderPropertiesAccessor properties) {
AbstractBindingPropertiesAccessor properties) {
return new Binding(name, moduleInputChannel, adapterFromBinder, CONSUMER, properties);
}
public static Binding forProducer(String name, MessageChannel moduleOutputChannel, AbstractEndpoint adapterToBinder,
AbstractBinderPropertiesAccessor properties) {
AbstractBindingPropertiesAccessor properties) {
return new Binding(name, moduleOutputChannel, adapterToBinder, PRODUCER, properties);
}
public static Binding forDirectProducer(String name, MessageChannel moduleOutputChannel,
AbstractEndpoint adapter, AbstractBinderPropertiesAccessor properties) {
AbstractEndpoint adapter, AbstractBindingPropertiesAccessor properties) {
return new Binding(name, moduleOutputChannel, adapter, DIRECT, properties);
}
@@ -91,7 +91,7 @@ public class Binding implements Lifecycle {
return type;
}
public AbstractBinderPropertiesAccessor getPropertiesAccessor() {
public AbstractBindingPropertiesAccessor getPropertiesAccessor() {
return properties;
}

44
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java
View File

@@ -123,42 +123,42 @@ public abstract class MessageChannelBinderSupport
*/
protected static final Set<Object> CONSUMER_STANDARD_PROPERTIES = new SetBuilder()
.add(BinderProperties.COUNT)
.add(BinderProperties.SEQUENCE)
.add(BinderPropertyKeys.COUNT)
.add(BinderPropertyKeys.SEQUENCE)
.build();
protected static final Set<Object> PRODUCER_STANDARD_PROPERTIES = new HashSet<Object>(Arrays.asList(
BinderProperties.NEXT_MODULE_COUNT,
BinderProperties.NEXT_MODULE_CONCURRENCY
BinderPropertyKeys.NEXT_MODULE_COUNT,
BinderPropertyKeys.NEXT_MODULE_CONCURRENCY
));
protected static final Set<Object> CONSUMER_RETRY_PROPERTIES = new HashSet<Object>(Arrays.asList(new String[] {
BinderProperties.BACK_OFF_INITIAL_INTERVAL,
BinderProperties.BACK_OFF_MAX_INTERVAL,
BinderProperties.BACK_OFF_MULTIPLIER,
BinderProperties.MAX_ATTEMPTS
BinderPropertyKeys.BACK_OFF_INITIAL_INTERVAL,
BinderPropertyKeys.BACK_OFF_MAX_INTERVAL,
BinderPropertyKeys.BACK_OFF_MULTIPLIER,
BinderPropertyKeys.MAX_ATTEMPTS
}));
protected static final Set<Object> PRODUCER_PARTITIONING_PROPERTIES = new HashSet<Object>(
Arrays.asList(new String[] {
BinderProperties.PARTITION_KEY_EXPRESSION,
BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS,
BinderProperties.PARTITION_SELECTOR_CLASS,
BinderProperties.PARTITION_SELECTOR_EXPRESSION,
BinderProperties.MIN_PARTITION_COUNT
BinderPropertyKeys.PARTITION_KEY_EXPRESSION,
BinderPropertyKeys.PARTITION_KEY_EXTRACTOR_CLASS,
BinderPropertyKeys.PARTITION_SELECTOR_CLASS,
BinderPropertyKeys.PARTITION_SELECTOR_EXPRESSION,
BinderPropertyKeys.MIN_PARTITION_COUNT
}));
protected static final Set<Object> PRODUCER_BATCHING_BASIC_PROPERTIES = new HashSet<Object>(
Arrays.asList(new String[] {
BinderProperties.BATCHING_ENABLED,
BinderProperties.BATCH_SIZE,
BinderProperties.BATCH_TIMEOUT,
BinderPropertyKeys.BATCHING_ENABLED,
BinderPropertyKeys.BATCH_SIZE,
BinderPropertyKeys.BATCH_TIMEOUT,
}));
protected static final Set<Object> PRODUCER_BATCHING_ADVANCED_PROPERTIES = new HashSet<Object>(
Arrays.asList(new String[] {
BinderProperties.BATCH_BUFFER_LIMIT,
BinderPropertyKeys.BATCH_BUFFER_LIMIT,
}));
private final List<Binding> bindings = Collections.synchronizedList(new ArrayList<Binding>());
@@ -802,7 +802,7 @@ public abstract class MessageChannelBinderSupport
* @param properties The properties.
* @return The retry template, or null if retry is not enabled.
*/
protected RetryTemplate buildRetryTemplateIfRetryEnabled(AbstractBinderPropertiesAccessor properties) {
protected RetryTemplate buildRetryTemplateIfRetryEnabled(AbstractBindingPropertiesAccessor properties) {
int maxAttempts = properties.getMaxAttempts(this.defaultMaxAttempts);
if (maxAttempts > 1) {
RetryTemplate template = new RetryTemplate();
@@ -835,7 +835,7 @@ public abstract class MessageChannelBinderSupport
* @return true if the producer is bound.
*/
protected boolean bindNewProducerDirectlyIfPossible(String name, SubscribableChannel moduleOutputChannel,
AbstractBinderPropertiesAccessor properties) {
AbstractBindingPropertiesAccessor properties) {
if (!properties.isDirectBindingAllowed()) {
return false;
}
@@ -868,7 +868,7 @@ public abstract class MessageChannelBinderSupport
}
private void bindProducerDirectly(String name, SubscribableChannel producerChannel,
MessageChannel consumerChannel, AbstractBinderPropertiesAccessor properties) {
MessageChannel consumerChannel, AbstractBindingPropertiesAccessor properties) {
DirectHandler handler = new DirectHandler(consumerChannel);
EventDrivenConsumer consumer = new EventDrivenConsumer(producerChannel, handler);
consumer.setBeanFactory(getBeanFactory());
@@ -899,7 +899,7 @@ public abstract class MessageChannelBinderSupport
}
}
if (producerBinding != null && producerBinding.getChannel() instanceof SubscribableChannel) {
AbstractBinderPropertiesAccessor properties = producerBinding.getPropertiesAccessor();
AbstractBindingPropertiesAccessor properties = producerBinding.getPropertiesAccessor();
if (properties.isDirectBindingAllowed()) {
bindProducerDirectly(name, (SubscribableChannel) producerBinding.getChannel(), consumerChannel,
properties);
@@ -969,7 +969,7 @@ public abstract class MessageChannelBinderSupport
private final int partitionCount;
public PartitioningMetadata(AbstractBinderPropertiesAccessor properties, int partitionCount) {
public PartitioningMetadata(AbstractBindingPropertiesAccessor properties, int partitionCount) {
this.partitionCount = partitionCount;
this.partitionKeyExtractorClass = properties.getPartitionKeyExtractorClass();
this.partitionKeyExpression = properties.getPartitionKeyExpression();

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

@@ -33,11 +33,27 @@ public class BindingProperties {
private static final String COMMA = ",";
/**
* The physical name at the broker that the binder binds to.
*/
private String destination;
private boolean partitioned = false;
/**
* Unique name that the binding belongs to.
*/
private String group = UUID.randomUUID().toString();
private int partitionCount = 1;
// Properties for both inbound/outbound
private String contentType;
private String binder;
private Boolean trackHistory;
// Outbound properties
// Partition properties
private String partitionKeyExpression;
@@ -47,13 +63,31 @@ public class BindingProperties {
private String partitionSelectorExpression;
private String group = UUID.randomUUID().toString();
private Integer partitionCount = 1;
private String contentType;
private Integer nextModuleCount;
private String binder;
private Integer nextModuleConcurrency;
// Batching properties
private Boolean batchingEnabled;
private Integer batchSize;
private Integer batchBufferLimit;
private Integer batchTimeout;
// Inbound properties
private Integer concurrency;
// Partition properties
private String partitionIndex;
private Boolean partitioned = false;
private boolean trackHistory;
public String getDestination() {
return this.destination;
@@ -63,20 +97,36 @@ public class BindingProperties {
this.destination = destination;
}
public boolean isPartitioned() {
return this.partitioned;
public String getGroup() {
return group;
}
public void setPartitioned(boolean partitioned) {
this.partitioned = partitioned;
public void setGroup(String group) {
this.group = group;
}
public int getPartitionCount() {
return this.partitionCount;
public String getContentType() {
return this.contentType;
}
public void setPartitionCount(int partitionCount) {
this.partitionCount = partitionCount;
public void setContentType(String contentType) {
this.contentType = contentType;
}
public String getBinder() {
return binder;
}
public void setBinder(String binder) {
this.binder = binder;
}
public Boolean getTrackHistory() {
return this.trackHistory;
}
public void setTrackHistory(Boolean trackHistory) {
this.trackHistory = trackHistory;
}
public String getPartitionKeyExpression() {
@@ -111,53 +161,104 @@ public class BindingProperties {
this.partitionSelectorExpression = partitionSelectorExpression;
}
public String getGroup() {
return group;
public Integer getNextModuleCount() {
return this.nextModuleCount;
}
public void setGroup(String group) {
this.group = group;
public void setNextModuleCount(Integer nextModuleCount) {
this.nextModuleCount = nextModuleCount;
}
public String getContentType() {
return this.contentType;
public Integer getNextModuleConcurrency() {
return this.nextModuleConcurrency;
}
public void setContentType(String contentType) {
this.contentType = contentType;
public void setNextModuleConcurrency(Integer nextModuleConcurrency) {
this.nextModuleConcurrency = nextModuleConcurrency;
}
public String getBinder() {
return binder;
public Boolean getBatchingEnabled() {
return this.batchingEnabled;
}
public void setBinder(String binder) {
this.binder = binder;
public void setBatchingEnabled(Boolean batchingEnabled) {
this.batchingEnabled = batchingEnabled;
}
public Boolean getTrackHistory() {
return this.trackHistory;
public Integer getBatchSize() {
return this.batchSize;
}
public void setTrackHistory(boolean trackHistory) {
this.trackHistory = trackHistory;
public void setBatchSize(Integer batchSize) {
this.batchSize = batchSize;
}
public Integer getBatchBufferLimit() {
return this.batchBufferLimit;
}
public void setBatchBufferLimit(Integer batchBufferLimit) {
this.batchBufferLimit = batchBufferLimit;
}
public Integer getBatchTimeout() {
return this.batchTimeout;
}
public void setBatchTimeout(Integer batchTimeout) {
this.batchTimeout = batchTimeout;
}
public Integer getPartitionCount() {
return this.partitionCount;
}
public void setPartitionCount(Integer partitionCount) {
this.partitionCount = partitionCount;
}
public Integer getConcurrency() {
return this.concurrency;
}
public void setConcurrency(Integer concurrency) {
this.concurrency = concurrency;
}
public String getPartitionIndex() {
return this.partitionIndex;
}
public void setPartitionIndex(String partitionIndex) {
this.partitionIndex = partitionIndex;
}
public Boolean getPartitioned() {
return this.partitioned;
}
public void setPartitioned(Boolean partitioned) {
this.partitioned = partitioned;
}
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("destination=" + destination);
sb.append("destination=" + this.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("group=" + this.group);
sb.append(COMMA);
if (this.contentType != null) {
sb.append("contentType=" + this.contentType);
sb.append(COMMA);
}
if (this.binder != null) {
sb.append("binder=" + this.binder);
sb.append(COMMA);
}
if (this.trackHistory != null) {
sb.append("trackHistory=" + this.trackHistory);
sb.append(COMMA);
}
if (this.partitionKeyExpression != null && !this.partitionKeyExpression.isEmpty()) {
sb.append("partitionKeyExpression=" + partitionKeyExpression);
sb.append(COMMA);
@@ -172,6 +273,45 @@ public class BindingProperties {
}
if (this.partitionSelectorClass != null && !this.partitionSelectorClass.isEmpty()) {
sb.append("partitionSelectorExpression=" + partitionSelectorExpression);
sb.append(COMMA);
}
if (this.partitioned) {
sb.append("partitionCount=" + this.partitionCount);
sb.append(COMMA);
}
if (this.nextModuleCount != null) {
sb.append("nextModuleCount=" + this.nextModuleCount);
sb.append(COMMA);
}
if (this.nextModuleConcurrency != null) {
sb.append("nextModuleConcurrency=" + this.nextModuleConcurrency);
sb.append(COMMA);
}
if (this.batchingEnabled != null) {
sb.append("batchingEnabled=" + this.batchingEnabled);
sb.append(COMMA);
}
if (this.batchSize != null) {
sb.append("batchSize=" + this.batchSize);
sb.append(COMMA);
}
if (this.batchBufferLimit != null) {
sb.append("batchBufferLimit=" + this.batchBufferLimit);
sb.append(COMMA);
}
if (this.batchTimeout != null) {
sb.append("batchTimeout=" + this.batchTimeout);
sb.append(COMMA);
}
sb.append("partitioned=" + this.partitioned);
sb.append(COMMA);
if (this.partitionIndex != null) {
sb.append("partitionIndex=" + this.partitionIndex);
sb.append(COMMA);
}
if (this.concurrency != null) {
sb.append("concurrency=" + this.concurrency);
sb.append(COMMA);
}
sb.deleteCharAt(sb.lastIndexOf(COMMA));
return "BindingProperties{" + sb.toString() + "}";

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

@@ -23,6 +23,7 @@ import java.util.TreeMap;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.util.StringUtils;
import com.fasterxml.jackson.annotation.JsonInclude;
@@ -32,6 +33,7 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include;
* @author Dave Syer
* @author Marius Bogoevici
* @author Gary Russell
* @author Ilayaperumal Gopinathan
*/
@ConfigurationProperties("spring.cloud.stream")
@JsonInclude(Include.NON_DEFAULT)
@@ -42,32 +44,12 @@ public class ChannelBindingServiceProperties {
private int instanceCount = 1;
private Properties consumerProperties = new Properties();
private Properties producerProperties = new Properties();
private Map<String, BindingProperties> bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
private Map<String, BinderProperties> binders = new HashMap<>();
private String defaultBinder;
public Properties getConsumerProperties() {
return this.consumerProperties;
}
public void setConsumerProperties(Properties consumerProperties) {
this.consumerProperties = consumerProperties;
}
public Properties getProducerProperties() {
return this.producerProperties;
}
public void setProducerProperties(Properties producerProperties) {
this.producerProperties = producerProperties;
}
public Map<String, BindingProperties> getBindings() {
return bindings;
}
@@ -116,80 +98,112 @@ public class ChannelBindingServiceProperties {
bindingProperties.getDestination() : channelName;
}
public int getPartitionCount(String channelName) {
return bindings.containsKey(channelName) ? bindings.get(channelName).getPartitionCount() : 1;
}
public boolean isPartitionedConsumer(String channelName) {
return bindings.containsKey(channelName) && bindings.get(channelName).isPartitioned();
}
public boolean isPartitionedProducer(String channelName) {
BindingProperties bindingProperties = bindings.get(channelName);
return bindingProperties != null &&
(StringUtils.hasText(bindingProperties.getPartitionKeyExpression())
|| StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass()));
}
/**
* Merge general properties provided by 'spring.cloud.stream.consumerProperties.*' with individual binding
* properties supplied via binders.
* Get consumer properties for the given input channel name.
*
* @param inputChannelName the input channel name
* @return merged consumer properties
*/
public Properties getConsumerProperties(String inputChannelName) {
if (isPartitionedConsumer(inputChannelName)) {
Properties channelConsumerProperties = new Properties();
channelConsumerProperties.putAll(consumerProperties);
channelConsumerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.COUNT,
Integer.toString(getInstanceCount()));
channelConsumerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_INDEX,
Integer.toString(getInstanceIndex()));
return channelConsumerProperties;
}
else {
return getConsumerProperties();
Properties channelConsumerProperties = new Properties();
BindingProperties bindingProperties = this.bindings.get(inputChannelName);
if (bindingProperties != null) {
if (bindingProperties.getConcurrency() != null) {
channelConsumerProperties.setProperty(BinderPropertyKeys.CONCURRENCY,
Integer.toString(bindingProperties.getConcurrency()));
}
updateConsumerPartitionProperties(inputChannelName, channelConsumerProperties);
}
return channelConsumerProperties;
}
/**
* Merge general properties provided by 'spring.cloud.stream.producerProperties.*' with individual binding
* properties supplied via binders.
* Get producer properties for the given output channel name.
*
* @param outputChannelName the output channel name
* @return merged producer properties
*/
public Properties getProducerProperties(String outputChannelName) {
if (isPartitionedProducer(outputChannelName)) {
Properties channelProducerProperties = new Properties();
channelProducerProperties.putAll(this.producerProperties);
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.NEXT_MODULE_COUNT,
Integer.toString(getPartitionCount(outputChannelName)));
BindingProperties bindingProperties = bindings.get(outputChannelName);
if (bindingProperties != null) {
Properties channelProducerProperties = new Properties();
updateBatchProperties(outputChannelName, channelProducerProperties);
updateProducerPartitionProperties(outputChannelName, channelProducerProperties);
return channelProducerProperties;
}
private boolean isPartitionedConsumer(String channelName) {
BindingProperties bindingProperties = bindings.get(channelName);
return bindingProperties != null && bindingProperties.getPartitioned();
}
private boolean isPartitionedProducer(String channelName) {
BindingProperties bindingProperties = bindings.get(channelName);
return (bindingProperties != null && (StringUtils.hasText(bindingProperties.getPartitionKeyExpression())
|| StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass())));
}
private void updateBatchProperties(String outputChannelName, Properties producerProperties) {
BindingProperties bindingProperties = this.bindings.get(outputChannelName);
if (bindingProperties != null) {
if (bindingProperties.getBatchingEnabled() != null) {
producerProperties.setProperty(BinderPropertyKeys.BATCHING_ENABLED,
String.valueOf(bindingProperties.getBatchingEnabled()));
}
if (bindingProperties.getBatchSize() != null) {
producerProperties.setProperty(BinderPropertyKeys.BATCH_SIZE,
String.valueOf(bindingProperties.getBatchSize()));
}
if (bindingProperties.getBatchBufferLimit() != null) {
producerProperties.setProperty(BinderPropertyKeys.BATCH_BUFFER_LIMIT,
String.valueOf(bindingProperties.getBatchBufferLimit()));
}
if (bindingProperties.getBatchTimeout() != null) {
producerProperties.setProperty(BinderPropertyKeys.BATCH_TIMEOUT,
String.valueOf(bindingProperties.getBatchTimeout()));
}
}
}
private void updateProducerPartitionProperties(String outputChannelName, Properties producerProperties) {
BindingProperties bindingProperties = this.bindings.get(outputChannelName);
if (bindingProperties != null) {
if (isPartitionedProducer(outputChannelName)) {
if (bindingProperties.getPartitionKeyExpression() != null) {
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_KEY_EXPRESSION,
producerProperties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXPRESSION,
bindingProperties.getPartitionKeyExpression());
}
if (bindingProperties.getPartitionKeyExtractorClass() != null) {
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS,
producerProperties.setProperty(BinderPropertyKeys.PARTITION_KEY_EXTRACTOR_CLASS,
bindingProperties.getPartitionKeyExtractorClass());
}
if (bindingProperties.getPartitionSelectorClass() != null) {
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_SELECTOR_CLASS,
producerProperties.setProperty(BinderPropertyKeys.PARTITION_SELECTOR_CLASS,
bindingProperties.getPartitionSelectorClass());
}
if (bindingProperties.getPartitionSelectorExpression() != null) {
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_SELECTOR_EXPRESSION,
producerProperties.setProperty(BinderPropertyKeys.PARTITION_SELECTOR_EXPRESSION,
bindingProperties.getPartitionSelectorExpression());
}
if (bindingProperties.getPartitionCount() != null) {
producerProperties.setProperty(BinderPropertyKeys.NEXT_MODULE_COUNT,
Integer.toString(bindingProperties.getPartitionCount()));
}
if (bindingProperties.getNextModuleConcurrency() != null) {
producerProperties.setProperty(BinderPropertyKeys.NEXT_MODULE_CONCURRENCY,
Integer.toString(bindingProperties.getNextModuleConcurrency()));
}
}
return channelProducerProperties;
}
else {
return this.producerProperties;
}
private void updateConsumerPartitionProperties(String inputChannelName, Properties consumerProperties) {
BindingProperties bindingProperties = this.bindings.get(inputChannelName);
if (bindingProperties != null) {
if (isPartitionedConsumer(inputChannelName)) {
consumerProperties.setProperty(BinderPropertyKeys.COUNT,
Integer.toString(getInstanceCount()));
consumerProperties.setProperty(BinderPropertyKeys.PARTITION_INDEX,
Integer.toString(getInstanceIndex()));
}
}
}
@@ -208,17 +222,14 @@ public class ChannelBindingServiceProperties {
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));
}
// Add Bindings properties
for (Map.Entry<String, BindingProperties> entry : getBindings().entrySet()) {
properties.put(entry.getKey(), entry.getValue().toString());
}
// Add Binder config properties
for (Map.Entry<String, BinderProperties> entry : binders.entrySet()) {
properties.put(entry.getKey(), entry.getValue());
}
return properties;
}

20
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java
View File

@@ -23,8 +23,8 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.cloud.stream.binder.AbstractBinderPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.AbstractBindingPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.integration.channel.DirectChannel;
@@ -72,7 +72,7 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
protected static final Set<Object> CONSUMER_REQUEST_REPLY_PROPERTIES = new SetBuilder()
.addAll(CONSUMER_STANDARD_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.add(BinderPropertyKeys.CONCURRENCY)
.build();
public static final String THREAD_NAME_PREFIX = "binder.local-";
@@ -237,7 +237,7 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel(name);
bridge(name, registeredChannel, moduleInputChannel,
"inbound." + ((NamedComponent) registeredChannel).getComponentName(),
new LocalBinderPropertiesAccessor(properties));
new LocalBindingPropertiesAccessor(properties));
}
/**
@@ -264,7 +264,7 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel(name);
bridge(name, moduleOutputChannel, registeredChannel,
"outbound." + ((NamedComponent) registeredChannel).getComponentName(),
new LocalBinderPropertiesAccessor(properties));
new LocalBindingPropertiesAccessor(properties));
}
@Override
@@ -331,7 +331,7 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
private ThreadPoolTaskExecutor createRequestReplyExecutor(String name, Properties properties) {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(new LocalBinderPropertiesAccessor(properties).getConcurrency(DEFAULT_REQ_REPLY_CONCURRENCY));
executor.setCorePoolSize(new LocalBindingPropertiesAccessor(properties).getConcurrency(DEFAULT_REQ_REPLY_CONCURRENCY));
executor.setThreadNamePrefix(THREAD_NAME_PREFIX + name + "-");
executor.initialize();
return executor;
@@ -351,13 +351,13 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
}
protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName,
LocalBinderPropertiesAccessor properties) {
LocalBindingPropertiesAccessor properties) {
return bridge(name, from, to, bridgeName, null, properties);
}
protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName,
final Collection<MimeType> acceptedMimeTypes, LocalBinderPropertiesAccessor properties) {
final Collection<MimeType> acceptedMimeTypes, LocalBindingPropertiesAccessor properties) {
final boolean isInbound = bridgeName.startsWith("inbound.");
@@ -412,9 +412,9 @@ public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
return getApplicationContext().getBean(name, requiredType);
}
private static class LocalBinderPropertiesAccessor extends AbstractBinderPropertiesAccessor {
private static class LocalBindingPropertiesAccessor extends AbstractBindingPropertiesAccessor {
public LocalBinderPropertiesAccessor(Properties properties) {
public LocalBindingPropertiesAccessor(Properties properties) {
super(properties);
}

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java
View File

@@ -35,7 +35,7 @@ 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.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
@@ -62,8 +62,8 @@ public class PartitionedConsumerTest {
public void testBindingPartitionedConsumer() {
ArgumentCaptor<Properties> argumentCaptor = ArgumentCaptor.forClass(Properties.class);
verify(binder).bindConsumer(eq("partIn"), eq(testSink.input()), argumentCaptor.capture());
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.PARTITION_INDEX), equalTo("0"));
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.COUNT),
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderPropertyKeys.PARTITION_INDEX), equalTo("0"));
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderPropertyKeys.COUNT),
equalTo("2"));
verifyNoMoreInteractions(binder);
}

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java
View File

@@ -35,7 +35,7 @@ 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.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.BinderPropertyKeys;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
@@ -62,8 +62,8 @@ public class PartitionedProducerTest {
public void testBindingPartitionedProducer() {
ArgumentCaptor<Properties> argumentCaptor = ArgumentCaptor.forClass(Properties.class);
verify(binder).bindProducer(eq("partOut"), eq(testSource.output()), argumentCaptor.capture());
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.NEXT_MODULE_COUNT), equalTo("3"));
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderProperties.PARTITION_KEY_EXPRESSION),
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderPropertyKeys.NEXT_MODULE_COUNT), equalTo("3"));
Assert.assertThat(argumentCaptor.getValue().getProperty(BinderPropertyKeys.PARTITION_KEY_EXPRESSION),
equalTo("payload"));
verifyNoMoreInteractions(binder);
}