From fa89e5bcb6bfc808015dce4c884a1ff06e990e5b Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Tue, 22 Dec 2015 22:41:28 +0530 Subject: [PATCH] Handle binder config in BindingProperties - 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 --- .../kafka/KafkaMessageChannelBinder.java | 22 +- .../stream/binder/kafka/KafkaBinderTests.java | 38 +-- .../binder/kafka/RawModeKafkaBinderTests.java | 10 +- .../rabbit/RabbitMessageChannelBinder.java | 24 +- .../binder/rabbit/RabbitBinderTests.java | 8 +- .../redis/RedisMessageChannelBinder.java | 16 +- .../stream/binder/redis/RedisBinderTests.java | 6 +- .../stream/binder/AbstractBinderTests.java | 2 +- .../stream/binder/BrokerBinderTests.java | 2 +- .../binder/PartitionCapableBinderTests.java | 8 +- .../config/MessageChannelConfigurerTests.java | 6 +- .../sink-channel-configurers.properties | 2 +- .../source-channel-configurers.properties | 2 +- ...=> AbstractBindingPropertiesAccessor.java} | 48 ++-- ...roperties.java => BinderPropertyKeys.java} | 3 +- .../cloud/stream/binder/Binding.java | 12 +- .../binder/MessageChannelBinderSupport.java | 44 ++-- .../stream/config/BindingProperties.java | 222 ++++++++++++++---- .../ChannelBindingServiceProperties.java | 157 +++++++------ .../local/LocalMessageChannelBinder.java | 20 +- .../partitioning/PartitionedConsumerTest.java | 6 +- .../partitioning/PartitionedProducerTest.java | 6 +- 22 files changed, 408 insertions(+), 256 deletions(-) rename spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/{AbstractBinderPropertiesAccessor.java => AbstractBindingPropertiesAccessor.java} (84%) rename spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/{BinderProperties.java => BinderPropertyKeys.java} (98%) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index bfdad178f..7e689b392 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -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 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 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 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 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); } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 249172a09..c092ad7cd 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -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); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index 0c8372740..38d0cd092 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -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"); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 064432040..8b7bd8fcd 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -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 RABBIT_CONSUMER_PROPERTIES = new HashSet(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 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 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 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 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 SUPPORTED_PUBSUB_PRODUCER_PROPERTIES = new SetBuilder() @@ -212,7 +212,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl private static final Set 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). diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 1042f2ed5..757fbcdb1 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java index 1bbaff09f..e9eb6f7a3 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java @@ -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 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 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 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 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); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java index bba61a312..50f052f57 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java @@ -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); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index 48c5503e4..36ffa8950 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -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 binder = getBinder(); try { binder.bindDynamicProducer("queue:foo", properties); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/BrokerBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/BrokerBinderTests.java index ff8a3f24d..60e03ceab 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/BrokerBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/BrokerBinderTests.java @@ -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"); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java index 9e6495945..d3e8262af 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java @@ -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"); diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index c5ead9894..a618405a9 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java @@ -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(); } }; 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 index 67edfec91..2ea698490 100644 --- 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 @@ -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 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 index 7a2b1b5bd..cd45020c2 100644 --- 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 @@ -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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderPropertiesAccessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java similarity index 84% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderPropertiesAccessor.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java index 65b2a7bd0..c0172bb06 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderPropertiesAccessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBindingPropertiesAccessor.java @@ -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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java similarity index 98% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderProperties.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java index 15f170def..10fbf1b61 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderPropertyKeys.java @@ -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. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index 1c5e8a2fc..aa653f983 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -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; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java index 1dc601caa..08027e54b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java @@ -123,42 +123,42 @@ public abstract class MessageChannelBinderSupport */ protected static final Set CONSUMER_STANDARD_PROPERTIES = new SetBuilder() - .add(BinderProperties.COUNT) - .add(BinderProperties.SEQUENCE) + .add(BinderPropertyKeys.COUNT) + .add(BinderPropertyKeys.SEQUENCE) .build(); protected static final Set PRODUCER_STANDARD_PROPERTIES = new HashSet(Arrays.asList( - BinderProperties.NEXT_MODULE_COUNT, - BinderProperties.NEXT_MODULE_CONCURRENCY + BinderPropertyKeys.NEXT_MODULE_COUNT, + BinderPropertyKeys.NEXT_MODULE_CONCURRENCY )); protected static final Set CONSUMER_RETRY_PROPERTIES = new HashSet(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 PRODUCER_PARTITIONING_PROPERTIES = new HashSet( 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 PRODUCER_BATCHING_BASIC_PROPERTIES = new HashSet( 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 PRODUCER_BATCHING_ADVANCED_PROPERTIES = new HashSet( Arrays.asList(new String[] { - BinderProperties.BATCH_BUFFER_LIMIT, + BinderPropertyKeys.BATCH_BUFFER_LIMIT, })); private final List bindings = Collections.synchronizedList(new ArrayList()); @@ -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(); 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 e253f87db..b4408e243 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 @@ -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() + "}"; 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 bd723336e..7fa136c94 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 @@ -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 bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); private Map 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 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 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 entry : getBindings().entrySet()) { properties.put(entry.getKey(), entry.getValue().toString()); } + // Add Binder config properties + for (Map.Entry entry : binders.entrySet()) { + properties.put(entry.getKey(), entry.getValue()); + } return properties; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java index 381353805..83de44c85 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java @@ -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 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 acceptedMimeTypes, LocalBinderPropertiesAccessor properties) { + final Collection 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); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java index 33a5b510c..6aed10986 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java @@ -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 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); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java index 09ff396e5..895549f67 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java @@ -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 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); }