From cadfd48a1f144c53e0966127beba444086891136 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Sun, 20 Mar 2016 17:24:24 -0400 Subject: [PATCH] Adds support for extended binding properties * Add parameterized ExtendedConsumerProperties and ExtendedProducerProperties; * Added ExtendedPropertiesBinder with the ability of managing per-binding property extensions and interacting the core service; * Add extensions for Kafka and Rabbit Addressing PR comments Changed binder-specific binding prefix to `spring.cloud.stream.binderType.bindings` Removing kafka-binder.properties --- .../binder/kafka/KafkaBindingProperties.java | 43 +++++ .../binder/kafka/KafkaConsumerProperties.java | 4 +- .../kafka/KafkaExtendedBindingProperties.java | 60 +++++++ .../kafka/KafkaMessageChannelBinder.java | 65 ++++--- .../binder/kafka/KafkaProducerProperties.java | 3 +- .../config/KafkaBinderConfiguration.java | 9 +- .../KafkaBinderConfigurationProperties.java | 27 +-- .../kafka-binder.properties | 18 -- .../stream/binder/kafka/KafkaBinderTests.java | 70 ++++---- .../stream/binder/kafka/KafkaTestBinder.java | 10 +- .../binder/kafka/RawModeKafkaBinderTests.java | 24 +-- .../rabbit/RabbitBindingProperties.java | 43 +++++ .../rabbit/RabbitConsumerProperties.java | 3 +- .../RabbitExtendedBindingProperties.java | 60 +++++++ .../rabbit/RabbitMessageChannelBinder.java | 97 +++++++---- .../rabbit/RabbitProducerProperties.java | 3 +- .../RabbitBinderConfigurationProperties.java | 2 +- ...bbitMessageChannelBinderConfiguration.java | 7 +- .../binder/rabbit/RabbitBinderTests.java | 164 +++++++++--------- .../binder/rabbit/RabbitTestBinder.java | 18 +- .../integration/RabbitBinderModuleTests.java | 45 +++++ .../binder/ExtendedBindingProperties.java | 30 ++++ .../binder/ExtendedConsumerProperties.java | 35 ++++ .../binder/ExtendedProducerProperties.java | 34 ++++ .../binder/ExtendedPropertiesBinder.java | 29 ++++ .../binding/BinderAwareChannelResolver.java | 5 +- .../stream/binding/ChannelBindingService.java | 70 +++----- .../stream/config/BindingProperties.java | 33 ++++ .../ChannelBindingServiceProperties.java | 60 ++----- .../stream/binder/DefaultSettingsTests.java | 130 -------------- .../PropertiesClassResolutionTests.java | 161 ----------------- .../partitioned-consumer-test.properties | 2 +- .../partitioned-producer-test.properties | 4 +- 33 files changed, 732 insertions(+), 636 deletions(-) create mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingProperties.java create mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java delete mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties create mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java create mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedBindingProperties.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedConsumerProperties.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedProducerProperties.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinder.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultSettingsTests.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/PropertiesClassResolutionTests.java diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingProperties.java new file mode 100644 index 000000000..6bd233d63 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingProperties.java @@ -0,0 +1,43 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka; + +/** + * @author Marius Bogoevici + */ +public class KafkaBindingProperties { + + private KafkaConsumerProperties consumer = new KafkaConsumerProperties(); + + private KafkaProducerProperties producer = new KafkaProducerProperties(); + + public KafkaConsumerProperties getConsumer() { + return consumer; + } + + public void setConsumer(KafkaConsumerProperties consumer) { + this.consumer = consumer; + } + + public KafkaProducerProperties getProducer() { + return producer; + } + + public void setProducer(KafkaProducerProperties producer) { + this.producer = producer; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java index 848bbc526..6894b3e70 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java @@ -16,12 +16,10 @@ package org.springframework.cloud.stream.binder.kafka; -import org.springframework.cloud.stream.binder.ConsumerProperties; - /** * @author Marius Bogoevici */ -public class KafkaConsumerProperties extends ConsumerProperties { +public class KafkaConsumerProperties { private int minPartitionCount = 1; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java new file mode 100644 index 000000000..637d14a35 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java @@ -0,0 +1,60 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.ExtendedBindingProperties; + +/** + * @author Marius Bogoevici + */ +@ConfigurationProperties("spring.cloud.stream.kafka") +public class KafkaExtendedBindingProperties implements ExtendedBindingProperties { + + private Map bindings = new HashMap<>(); + + public Map getBindings() { + return bindings; + } + + public void setBindings(Map bindings) { + this.bindings = bindings; + } + + @Override + public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) { + if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) { + return bindings.get(channelName).getConsumer(); + } + else { + return new KafkaConsumerProperties(); + } + } + + @Override + public KafkaProducerProperties getExtendedProducerProperties(String channelName) { + if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) { + return bindings.get(channelName).getProducer(); + } + else { + return new KafkaProducerProperties(); + } + } +} 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 2f393e21f..6c371e5b4 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 @@ -49,6 +49,9 @@ import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.DefaultBinding; import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.cloud.stream.binder.PartitionHandler; import org.springframework.http.MediaType; @@ -98,7 +101,7 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Soby Chacko */ -public class KafkaMessageChannelBinder extends AbstractBinder { +public class KafkaMessageChannelBinder extends AbstractBinder, ExtendedProducerProperties> implements ExtendedPropertiesBinder { public static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer(); @@ -147,6 +150,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder> getTopicsInUse() { return this.topicsInUse; } @Override - protected Binding doBindConsumer(String name, String group, MessageChannel inputChannel, KafkaConsumerProperties properties) { + protected Binding doBindConsumer(String name, String group, MessageChannel inputChannel, + ExtendedConsumerProperties properties) { // If the caller provides a consumer group, use it; otherwise an anonymous consumer group // is generated each time, such that each anonymous binding will receive all messages. // Consumers reset offsets at the latest time by default, which allows them to receive only @@ -300,14 +320,15 @@ public class KafkaMessageChannelBinder extends AbstractBinder doBindProducer(String name, MessageChannel moduleOutputChannel, KafkaProducerProperties properties) { + public Binding doBindProducer(String name, MessageChannel moduleOutputChannel, + ExtendedProducerProperties properties) { Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel); if (logger.isInfoEnabled()) { @@ -323,12 +344,12 @@ public class KafkaMessageChannelBinder extends AbstractBinder producerMetadata = new ProducerMetadata<>(name, byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); - producerMetadata.setSync(properties.isSync()); - producerMetadata.setCompressionType(properties.getCompressionType()); - producerMetadata.setBatchBytes(properties.getBufferSize()); + producerMetadata.setSync(properties.getExtension().isSync()); + producerMetadata.setCompressionType(properties.getExtension().getCompressionType()); + producerMetadata.setBatchBytes(properties.getExtension().getBufferSize()); Properties additionalProps = new Properties(); additionalProps.put(ProducerConfig.ACKS_CONFIG, String.valueOf(requiredAcks)); - additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(properties.getBatchTimeout())); + additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(properties.getExtension().getBatchTimeout())); ProducerFactoryBean producerFB = new ProducerFactoryBean<>(producerMetadata, brokers, additionalProps); try { @@ -406,10 +427,10 @@ public class KafkaMessageChannelBinder extends AbstractBinder createKafkaConsumer(String name, final MessageChannel moduleInputChannel, - KafkaConsumerProperties properties, String group, long referencePoint) { + ExtendedConsumerProperties properties, String group, long referencePoint) { validateTopicName(name); - int minKafkaPartitions = properties.getMinPartitionCount(); + int minKafkaPartitions = properties.getExtension().getMinPartitionCount(); int instance = properties.getInstanceCount(); if (instance == 0) { throw new IllegalArgumentException("Instance count cannot be zero"); @@ -450,7 +471,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder consumerProperties, String group, String topic, Collection listenedPartitions, long referencePoint) { Assert.isTrue(StringUtils.hasText(topic) ^ !CollectionUtils.isEmpty(listenedPartitions), @@ -500,7 +522,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder consumerProperties; - public ReceivingHandler(KafkaConsumerProperties consumerProperties) { + public ReceivingHandler(ExtendedConsumerProperties consumerProperties) { this.consumerProperties = consumerProperties; } @Override @SuppressWarnings("unchecked") protected Object handleRequestMessage(Message requestMessage) { - if (Mode.embeddedHeaders.equals(consumerProperties.getMode())) { + if (Mode.embeddedHeaders.equals(consumerProperties.getExtension().getMode())) { MessageValues messageValues; try { messageValues = embeddedHeadersMessageConverter.extractHeaders((Message) requestMessage, @@ -596,7 +618,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder producerProperties; private final int numberOfKafkaPartitions; @@ -604,7 +626,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder properties, + int numberOfPartitions, ProducerConfiguration producerConfiguration) { this.topicName = topicName; producerProperties = properties; @@ -626,13 +649,13 @@ public class KafkaMessageChannelBinder extends AbstractBinder { +public class KafkaBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); @@ -88,13 +90,13 @@ public class KafkaBinderTests extends PartitionCapableBinderTests createConsumerProperties() { + return new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); } @Override - protected KafkaProducerProperties createProducerProperties() { - return new KafkaProducerProperties(); + protected ExtendedProducerProperties createProducerProperties() { + return new ExtendedProducerProperties<>(new KafkaProducerProperties()); } @Before @@ -165,10 +167,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests producerProperties = createProducerProperties(); + producerProperties.getExtension().setCompressionType(codec); Binding producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties); - Binding consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, new KafkaConsumerProperties()); + Binding consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, createConsumerProperties()); Message message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build(); // Let the consumer actually bind to the producer before sending a msg binderBindUnbindLatency(); @@ -190,10 +192,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests producerProperties = createProducerProperties(); producerProperties.setPartitionCount(10); - KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties(); - consumerProperties.setMinPartitionCount(10); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setMinPartitionCount(10); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties); @@ -220,11 +222,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests producerProperties = createProducerProperties(); producerProperties.setPartitionCount(5); producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); - KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties(); - consumerProperties.setMinPartitionCount(3); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setMinPartitionCount(3); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties); @@ -251,11 +253,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests producerProperties = createProducerProperties(); producerProperties.setPartitionCount(5); producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); - KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties(); - consumerProperties.setMinPartitionCount(5); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setMinPartitionCount(5); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties); @@ -286,10 +288,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests(testPayload1.getBytes())); - binder.bindConsumer(testTopicName, "startOffsets", input1, new KafkaConsumerProperties()); + binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties()); Message receivedMessage1 = (Message) receive(input1); assertThat(receivedMessage1, not(nullValue())); assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1)); @@ -308,11 +310,11 @@ public class KafkaBinderTests extends PartitionCapableBinderTests(testPayload1.getBytes())); - KafkaConsumerProperties properties = new KafkaConsumerProperties(); - properties.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); binder.bindConsumer(testTopicName, "startOffsets", input1, properties); Message receivedMessage1 = (Message) receive(input1); assertThat(receivedMessage1, not(nullValue())); @@ -332,12 +334,12 @@ public class KafkaBinderTests extends PartitionCapableBinderTests producerBinding = binder.bindProducer(testTopicName, output, new KafkaProducerProperties()); + Binding producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties()); String testPayload1 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload1.getBytes())); - KafkaConsumerProperties properties = new KafkaConsumerProperties(); - properties.setResetOffsets(true); - properties.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.getExtension().setResetOffsets(true); + properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties); Message receivedMessage1 = (Message) receive(input1); @@ -352,9 +354,9 @@ public class KafkaBinderTests extends PartitionCapableBinderTests(testPayload3.getBytes())); - KafkaConsumerProperties properties2 = new KafkaConsumerProperties(); - properties2.setResetOffsets(true); - properties2.setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); + ExtendedConsumerProperties properties2 = createConsumerProperties(); + properties2.getExtension().setResetOffsets(true); + properties2.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties2); Message receivedMessage4 = (Message) receive(input1); @@ -383,10 +385,10 @@ public class KafkaBinderTests extends PartitionCapableBinderTests producerBinding = binder.bindProducer(testTopicName, output, new KafkaProducerProperties()); + Binding producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties()); String testPayload1 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload1.getBytes())); - KafkaConsumerProperties firstConsumerProperties = new KafkaConsumerProperties(); + ExtendedConsumerProperties firstConsumerProperties = createConsumerProperties(); Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, firstConsumerProperties); Message receivedMessage1 = (Message) receive(input1); assertThat(receivedMessage1, not(nullValue())); @@ -401,7 +403,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests(testPayload3.getBytes())); consumerBinding = - binder.bindConsumer(testTopicName, "startOffsets", input1, new KafkaConsumerProperties()); + binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties()); Message receivedMessage3 = (Message) receive(input1); assertThat(receivedMessage3, not(nullValue())); assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3)); @@ -421,8 +423,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests properties = createProducerProperties(); + properties.getExtension().setSync(true); Binding producerBinding = binder.bindProducer(testTopicName, output, properties); DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); MessageHandler handler = (MessageHandler) accessor.getPropertyValue("handler"); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 49563d988..29f0c5ba4 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -18,7 +18,12 @@ package org.springframework.cloud.stream.binder.kafka; import java.util.List; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Registration; + import org.springframework.cloud.stream.binder.AbstractTestBinder; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport; import org.springframework.cloud.stream.test.junit.kafka.TestKafkaCluster; import org.springframework.context.support.GenericApplicationContext; @@ -30,9 +35,6 @@ import org.springframework.integration.kafka.support.ProducerListener; import org.springframework.integration.kafka.support.ZookeeperConnect; import org.springframework.integration.tuple.TupleKryoRegistrar; -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.Registration; - /** * Test support class for {@link KafkaMessageChannelBinder}. @@ -43,7 +45,7 @@ import com.esotericsoftware.kryo.Registration; * @author Gary Russell * @author Soby Chacko */ -public class KafkaTestBinder extends AbstractTestBinder { +public class KafkaTestBinder extends AbstractTestBinder, ExtendedProducerProperties> { public KafkaTestBinder(KafkaTestSupport kafkaTestSupport) { 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 e66f55391..3f2910392 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 @@ -31,6 +31,8 @@ import org.junit.Test; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -52,7 +54,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Override public void testPartitionedModuleJava() throws Exception { KafkaTestBinder binder = getBinder(); - KafkaProducerProperties properties = new KafkaProducerProperties(); + ExtendedProducerProperties properties = createProducerProperties(); properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class); properties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class); properties.setPartitionCount(3); @@ -61,7 +63,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("partJ.0", output, properties); - KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setConcurrency(2); consumerProperties.setInstanceCount(3); consumerProperties.setInstanceIndex(0); @@ -105,7 +107,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Override public void testPartitionedModuleSpEL() throws Exception { KafkaTestBinder binder = getBinder(); - KafkaProducerProperties properties = new KafkaProducerProperties(); + ExtendedProducerProperties properties = createProducerProperties(); properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]")); properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); properties.setPartitionCount(3); @@ -121,7 +123,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { } - KafkaConsumerProperties consumerProperties = new KafkaConsumerProperties(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setConcurrency(2); consumerProperties.setInstanceIndex(0); consumerProperties.setInstanceCount(3); @@ -173,8 +175,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { KafkaTestBinder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); QueueChannel moduleInputChannel = new QueueChannel(); - Binding producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, new KafkaProducerProperties()); - Binding consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, new KafkaConsumerProperties()); + Binding producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, createConsumerProperties()); Message message = MessageBuilder.withPayload("foo".getBytes()).build(); // Let the consumer actually bind to the producer before sending a msg binderBindUnbindLatency(); @@ -202,14 +204,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { QueueChannel module1InputChannel = new QueueChannel(); QueueChannel module2InputChannel = new QueueChannel(); QueueChannel module3InputChannel = new QueueChannel(); - Binding producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, new KafkaProducerProperties()); - Binding input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel, new KafkaConsumerProperties()); + Binding producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, createProducerProperties()); + Binding input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel, createConsumerProperties()); // A new module is using the tap as an input channel String fooTapName = "baz.0"; - Binding input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel, new KafkaConsumerProperties()); + Binding input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel, createConsumerProperties()); // Another new module is using tap as an input channel String barTapName = "baz.0"; - Binding input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, new KafkaConsumerProperties()); + Binding input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, createConsumerProperties()); Message message = MessageBuilder.withPayload("foo".getBytes()).build(); boolean success = false; @@ -245,7 +247,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { assertNull(receive(module3InputChannel)); // re-subscribed tap does receive the message - input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, new KafkaConsumerProperties()); + input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, createConsumerProperties()); assertNotNull(receive(module3InputChannel)); // clean up diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java new file mode 100644 index 000000000..5ed04136d --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java @@ -0,0 +1,43 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.rabbit; + +/** + * @author Marius Bogoevici + */ +public class RabbitBindingProperties { + + private RabbitConsumerProperties consumer = new RabbitConsumerProperties(); + + private RabbitProducerProperties producer = new RabbitProducerProperties(); + + public RabbitConsumerProperties getConsumer() { + return consumer; + } + + public void setConsumer(RabbitConsumerProperties consumer) { + this.consumer = consumer; + } + + public RabbitProducerProperties getProducer() { + return producer; + } + + public void setProducer(RabbitProducerProperties producer) { + this.producer = producer; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java index 3a3775e7b..9a635cb42 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java @@ -17,13 +17,12 @@ package org.springframework.cloud.stream.binder.rabbit; import org.springframework.amqp.core.AcknowledgeMode; -import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.util.Assert; /** * @author Marius Bogoevici */ -public class RabbitConsumerProperties extends ConsumerProperties { +public class RabbitConsumerProperties { private String prefix = ""; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java new file mode 100644 index 000000000..afc821c4f --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java @@ -0,0 +1,60 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.rabbit; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.ExtendedBindingProperties; + +/** + * @author Marius Bogoevici + */ +@ConfigurationProperties("spring.cloud.stream.rabbit") +public class RabbitExtendedBindingProperties implements ExtendedBindingProperties { + + private Map bindings = new HashMap<>(); + + public Map getBindings() { + return bindings; + } + + public void setBindings(Map bindings) { + this.bindings = bindings; + } + + @Override + public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) { + if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) { + return bindings.get(channelName).getConsumer(); + } + else { + return new RabbitConsumerProperties(); + } + } + + @Override + public RabbitProducerProperties getExtendedProducerProperties(String channelName) { + if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) { + return bindings.get(channelName).getProducer(); + } + else { + return new RabbitProducerProperties(); + } + } +} 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 72ccfafe4..f6e8b6f9b 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 @@ -62,6 +62,9 @@ import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.DefaultBinding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.cloud.stream.binder.PartitionHandler; import org.springframework.context.Lifecycle; @@ -99,8 +102,8 @@ import org.springframework.util.StringUtils; * @author David Turanski * @author Marius Bogoevici */ -public class RabbitMessageChannelBinder extends AbstractBinder { +public class RabbitMessageChannelBinder extends AbstractBinder, + ExtendedProducerProperties> implements ExtendedPropertiesBinder { public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR = new AnonymousQueue.Base64UrlNamingStrategy("anonymous."); @@ -151,6 +154,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder doBindConsumer(String name, String group, MessageChannel inputChannel, - RabbitConsumerProperties properties) { + ExtendedConsumerProperties properties) { boolean anonymousConsumer = !StringUtils.hasText(group); String baseQueueName = anonymousConsumer ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) : groupedName(name, group); if (this.logger.isInfoEnabled()) { this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name); } - String prefix = properties.getPrefix(); + String prefix = properties.getExtension().getPrefix(); String exchangeName = applyPrefix(prefix, name); TopicExchange exchange = new TopicExchange(exchangeName); declareExchange(exchangeName, exchange); String queueName = applyPrefix(prefix, baseQueueName); boolean partitioned = !anonymousConsumer && properties.isPartitioned(); - boolean durable = !anonymousConsumer && properties.isDurableSubscription(); + boolean durable = !anonymousConsumer && properties.getExtension().isDurableSubscription(); Queue queue; if (anonymousConsumer) { @@ -254,7 +274,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder binding = doRegisterConsumer(baseQueueName, group, inputChannel, queue, properties); if (durable) { - autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getPrefix(), properties.isAutoBindDlq()); + autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()); } return binding; } @@ -287,34 +307,34 @@ public class RabbitMessageChannelBinder extends AbstractBinder doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue, - final RabbitConsumerProperties properties) { + final ExtendedConsumerProperties properties) { DefaultBinding consumerBinding; SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer( this.connectionFactory); - listenerContainer.setAcknowledgeMode(properties.getAcknowledgeMode()); - listenerContainer.setChannelTransacted(properties.isTransacted()); - listenerContainer.setDefaultRequeueRejected(properties.isRequeueRejected()); + listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode()); + listenerContainer.setChannelTransacted(properties.getExtension().isTransacted()); + listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected()); int concurrency = properties.getConcurrency(); concurrency = concurrency > 0 ? concurrency : 1; listenerContainer.setConcurrentConsumers(concurrency); - int maxConcurrency = properties.getMaxConcurrency(); + int maxConcurrency = properties.getExtension().getMaxConcurrency(); if (maxConcurrency > concurrency) { listenerContainer.setMaxConcurrentConsumers(maxConcurrency); } - listenerContainer.setPrefetchCount(properties.getPrefetch()); - listenerContainer.setTxSize(properties.getTxSize()); + listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch()); + listenerContainer.setTxSize(properties.getExtension().getTxSize()); listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-")); listenerContainer.setQueues(queue); int maxAttempts = properties.getMaxAttempts(); - if (maxAttempts > 1 || properties.isRepublishToDlq()) { + if (maxAttempts > 1 || properties.getExtension().isRepublishToDlq()) { RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless() .maxAttempts(maxAttempts) .backOffOptions(properties.getBackOffInitialInterval(), properties.getBackOffMultiplier(), properties.getBackOffMaxInterval()) - .recoverer(determineRecoverer(name, properties.getPrefix(), properties.isRepublishToDlq())) + .recoverer(determineRecoverer(name, properties.getExtension().getPrefix(), properties.getExtension().isRepublishToDlq())) .build(); listenerContainer.setAdviceChain(new Advice[] { retryInterceptor }); } @@ -329,14 +349,14 @@ public class RabbitMessageChannelBinder extends AbstractBinder(name, group, moduleInputChannel, adapter) { @Override protected void afterUnbind() { - cleanAutoDeclareContext(properties.getPrefix(), name); + cleanAutoDeclareContext(properties.getExtension().getPrefix(), name); } }; ReceivingHandler convertingBridge = new ReceivingHandler(); @@ -361,9 +381,10 @@ public class RabbitMessageChannelBinder extends AbstractBinder properties, RabbitTemplate rabbitTemplate) { - String prefix = properties.getPrefix(); + String prefix = properties.getExtension().getPrefix(); String exchangeName = applyPrefix(prefix, name); TopicExchange exchange = new TopicExchange(exchangeName); declareExchange(exchangeName, exchange); @@ -378,9 +399,9 @@ public class RabbitMessageChannelBinder extends AbstractBinder producerProperties) { DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper(); - mapper.setRequestHeaderNames(producerProperties.getRequestHeaderPatterns()); - mapper.setReplyHeaderNames(producerProperties.getReplyHeaderPatterns()); + mapper.setRequestHeaderNames(producerProperties.getExtension().getRequestHeaderPatterns()); + mapper.setReplyHeaderNames(producerProperties.getExtension().getReplyHeaderPatterns()); handler.setHeaderMapper(mapper); - handler.setDefaultDeliveryMode(producerProperties.getDeliveryMode()); + handler.setDefaultDeliveryMode(producerProperties.getExtension().getDeliveryMode()); handler.setBeanFactory(this.getBeanFactory()); handler.afterPropertiesSet(); } @Override - public Binding doBindProducer(String name, MessageChannel outputChannel, RabbitProducerProperties producerProperties) { - String exchangeName = applyPrefix(producerProperties.getPrefix(), name); + public Binding doBindProducer(String name, MessageChannel outputChannel, + ExtendedProducerProperties producerProperties) { + String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name); TopicExchange exchange = new TopicExchange(exchangeName); declareExchange(exchangeName, exchange); AmqpOutboundEndpoint endpoint = this.buildOutboundEndpoint(name, producerProperties, - buildRabbitTemplate(producerProperties)); + buildRabbitTemplate(producerProperties.getExtension())); return doRegisterProducer(name, outputChannel, endpoint, producerProperties); } @@ -445,12 +468,13 @@ public class RabbitMessageChannelBinder extends AbstractBinder doRegisterProducer(final String name, MessageChannel moduleOutputChannel, - AmqpOutboundEndpoint delegate, RabbitProducerProperties properties) { + AmqpOutboundEndpoint delegate, ExtendedProducerProperties properties) { return this.doRegisterProducer(name, moduleOutputChannel, delegate, null, properties); } private Binding doRegisterProducer(final String name, MessageChannel moduleOutputChannel, - AmqpOutboundEndpoint delegate, String replyTo, RabbitProducerProperties properties) { + AmqpOutboundEndpoint delegate, String replyTo, + ExtendedProducerProperties properties) { Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel); MessageHandler handler = new SendingHandler(delegate, replyTo, properties); EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler); @@ -589,11 +613,12 @@ public class RabbitMessageChannelBinder extends AbstractBinder producerProperties; private final PartitionHandler partitionHandler; - private SendingHandler(MessageHandler delegate, String replyTo, RabbitProducerProperties properties) { + private SendingHandler(MessageHandler delegate, String replyTo, + ExtendedProducerProperties properties) { this.delegate = delegate; this.replyTo = replyTo; producerProperties = properties; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java index bcd29b4ff..35f391844 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java @@ -17,13 +17,12 @@ package org.springframework.cloud.stream.binder.rabbit; import org.springframework.amqp.core.MessageDeliveryMode; -import org.springframework.cloud.stream.binder.ProducerProperties; /** * @author Marius Bogoevici * @author Gary Russell */ -public class RabbitProducerProperties extends ProducerProperties { +public class RabbitProducerProperties { private String prefix = ""; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java index db829be9c..5caad6304 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java @@ -22,7 +22,7 @@ import org.springframework.core.io.Resource; /** * @author David Turanski */ -@ConfigurationProperties(prefix = "spring.cloud.stream.binder.rabbit") +@ConfigurationProperties(prefix = "spring.cloud.stream.rabbit.binder") class RabbitBinderConfigurationProperties { private String[] addresses = new String[0]; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index 412be37c0..a06a116f8 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -28,6 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.rabbit.ConnectionFactorySettings; +import org.springframework.cloud.stream.binder.rabbit.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; import org.springframework.context.annotation.Bean; @@ -37,7 +38,7 @@ import org.springframework.integration.codec.Codec; @Configuration @Import({PropertyPlaceholderAutoConfiguration.class, KryoCodecAutoConfiguration.class}) -@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class}) +@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class}) public class RabbitMessageChannelBinderConfiguration { @Autowired @@ -49,6 +50,9 @@ public class RabbitMessageChannelBinderConfiguration { @Autowired private RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties; + @Autowired + private RabbitExtendedBindingProperties rabbitExtendedBindingProperties; + @Bean RabbitMessageChannelBinder rabbitMessageChannelBinder() { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory); @@ -63,6 +67,7 @@ public class RabbitMessageChannelBinderConfiguration { binder.setUsername(rabbitBinderConfigurationProperties.getUsername()); binder.setUseSSL(rabbitBinderConfigurationProperties.isUseSSL()); binder.setVhost(rabbitBinderConfigurationProperties.getVhost()); + binder.setExtendedBindingProperties(rabbitExtendedBindingProperties); return binder; } 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 3c7d0b581..874e9cfa9 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 @@ -55,6 +55,8 @@ import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPos import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; import org.springframework.cloud.stream.binder.PartitionSelectorStrategy; @@ -79,7 +81,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Gary Russell * @author David Turanski */ -public class RabbitBinderTests extends PartitionCapableBinderTests { +public class RabbitBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName(); @@ -97,13 +99,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests createConsumerProperties() { + return new ExtendedConsumerProperties<>(new RabbitConsumerProperties()); } @Override - protected RabbitProducerProperties createProducerProperties() { - return new RabbitProducerProperties(); + protected ExtendedProducerProperties createProducerProperties() { + return new ExtendedProducerProperties<>(new RabbitProducerProperties()); } @Override @@ -116,8 +118,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests producerBinding = binder.bindProducer("bad.0", moduleOutputChannel, new RabbitProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, new RabbitConsumerProperties()); + Binding producerBinding = binder.bindProducer("bad.0", moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, createConsumerProperties()); Message message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); final CountDownLatch latch = new CountDownLatch(3); @@ -138,15 +140,15 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createConsumerProperties(); + properties.getExtension().setTransacted(true); Binding consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties); AbstractEndpoint endpoint = extractEndpoint(consumerBinding); SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", SimpleMessageListenerContainer.class); assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode()); assertThat(container.getQueueNames()[0], - startsWith(properties.getPrefix())); + startsWith(properties.getExtension().getPrefix())); assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class)); assertEquals(1, TestUtils.getPropertyValue(container, "concurrentConsumers")); assertNull(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")); @@ -161,19 +163,19 @@ public class RabbitBinderTests extends PartitionCapableBinderTests producerBinding = binder.bindProducer("props.0", new DirectChannel(), new RabbitProducerProperties()); + Binding producerBinding = binder.bindProducer("props.0", new DirectChannel(), createProducerProperties()); @SuppressWarnings("unchecked") AbstractEndpoint endpoint = extractEndpoint(producerBinding); MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode", @@ -202,16 +204,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createProducerProperties(); + properties.getExtension().setPrefix("foo."); + properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + properties.getExtension().setRequestHeaderPatterns(new String[] {"foo"}); properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'")); properties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class); properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0")); properties.setPartitionSelectorClass(TestPartitionSelectorClass.class); properties.setPartitionCount(1); - properties.setTransacted(true); + properties.getExtension().setTransacted(true); producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties); endpoint = extractEndpoint(producerBinding); @@ -235,12 +237,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createConsumerProperties(); + properties.getExtension().setPrefix(TEST_PREFIX); + properties.getExtension().setAutoBindDlq(true); + properties.getExtension().setDurableSubscription(true); properties.setMaxAttempts(1); // disable retry - properties.setRequeueRejected(false); + properties.getExtension().setRequeueRejected(false); DirectChannel moduleInputChannel = new DirectChannel(); moduleInputChannel.setBeanName("durableTest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -276,12 +278,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createConsumerProperties(); + properties.getExtension().setPrefix(TEST_PREFIX); + properties.getExtension().setAutoBindDlq(true); + properties.getExtension().setDurableSubscription(false); properties.setMaxAttempts(1); // disable retry - properties.setRequeueRejected(false); + properties.getExtension().setRequeueRejected(false); DirectChannel moduleInputChannel = new DirectChannel(); moduleInputChannel.setBeanName("nondurabletest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -301,12 +303,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createConsumerProperties(); + properties.getExtension().setPrefix(TEST_PREFIX); + properties.getExtension().setAutoBindDlq(true); properties.setMaxAttempts(1); // disable retry - properties.setRequeueRejected(false); - properties.setDurableSubscription(true); + properties.getExtension().setRequeueRejected(false); + properties.getExtension().setDurableSubscription(true); DirectChannel moduleInputChannel = new DirectChannel(); moduleInputChannel.setBeanName("dlqTest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -346,11 +348,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createConsumerProperties(); + properties.getExtension().setPrefix("bindertest."); + properties.getExtension().setAutoBindDlq(true); properties.setMaxAttempts(1); // disable retry - properties.setRequeueRejected(false); + properties.getExtension().setRequeueRejected(false); properties.setPartitioned(true); properties.setInstanceIndex(0); DirectChannel input0 = new DirectChannel(); @@ -364,9 +366,9 @@ public class RabbitBinderTests extends PartitionCapableBinderTests input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties); Binding defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties); - RabbitProducerProperties producerProperties = new RabbitProducerProperties(); - producerProperties.setPrefix("bindertest."); - producerProperties.setAutoBindDlq(true); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().setPrefix("bindertest."); + producerProperties.getExtension().setAutoBindDlq(true); producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class); producerProperties.setPartitionSelectorClass(PartitionTestSupport.class); producerProperties.setPartitionCount(2); @@ -432,10 +434,10 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createProducerProperties(); - properties.setPrefix("bindertest."); - properties.setAutoBindDlq(true); + properties.getExtension().setPrefix("bindertest."); + properties.getExtension().setAutoBindDlq(true); properties.setRequiredGroups("dlqPartGrp"); properties.setPartitionKeyExtractorClass(PartitionTestSupport.class); properties.setPartitionSelectorClass(PartitionTestSupport.class); @@ -444,11 +446,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests outputBinding = binder.bindProducer("partDLQ.1", output, properties); - RabbitConsumerProperties consumerProperties = new RabbitConsumerProperties(); - consumerProperties.setPrefix("bindertest."); - consumerProperties.setAutoBindDlq(true); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix("bindertest."); + consumerProperties.getExtension().setAutoBindDlq(true); consumerProperties.setMaxAttempts(1); // disable retry - consumerProperties.setRequeueRejected(false); + consumerProperties.getExtension().setRequeueRejected(false); consumerProperties.setPartitioned(true); consumerProperties.setInstanceIndex(0); DirectChannel input0 = new DirectChannel(); @@ -527,13 +529,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createConsumerProperties(); + properties.getExtension().setPrefix(TEST_PREFIX); + properties.getExtension().setAutoBindDlq(true); + properties.getExtension().setRepublishToDlq(true); properties.setMaxAttempts(1); // disable retry - properties.setRequeueRejected(false); - properties.setDurableSubscription(true); + properties.getExtension().setRequeueRejected(false); + properties.getExtension().setDurableSubscription(true); DirectChannel moduleInputChannel = new DirectChannel(); moduleInputChannel.setBeanName("dlqPubTest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -569,20 +571,20 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createProducerProperties(); + properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + properties.getExtension().setBatchingEnabled(true); + properties.getExtension().setBatchSize(2); + properties.getExtension().setBatchBufferLimit(100000); + properties.getExtension().setBatchTimeout(30000); + properties.getExtension().setCompress(true); properties.setRequiredGroups("default"); DirectChannel output = new DirectChannel(); output.setBeanName("batchingProducer"); Binding producerBinding = binder.bindProducer("batching.0", output, properties); - while (template.receive(properties.getPrefix() + "batching.0.default") != null) { + while (template.receive(properties.getExtension().getPrefix() + "batching.0.default") != null) { } Log logger = spy(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.logger", Log.class)); @@ -605,7 +607,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests consumerBinding = binder.bindConsumer("batching.0", "test", input, new RabbitConsumerProperties()); + Binding consumerBinding = binder.bindConsumer("batching.0", "test", input, createConsumerProperties()); output.send(new GenericMessage<>("foo".getBytes())); output.send(new GenericMessage<>("bar".getBytes())); @@ -633,16 +635,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests properties = createProducerProperties(); + properties.getExtension().setPrefix("latebinder."); + properties.getExtension().setAutoBindDlq(true); MessageChannel moduleOutputChannel = new DirectChannel(); Binding late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, properties); QueueChannel moduleInputChannel = new QueueChannel(); - RabbitConsumerProperties rabbitConsumerProperties = new RabbitConsumerProperties(); - rabbitConsumerProperties.setPrefix("latebinder."); + ExtendedConsumerProperties rabbitConsumerProperties = createConsumerProperties(); + rabbitConsumerProperties.getExtension().setPrefix("latebinder."); Binding late0ConsumerBinding = binder.bindConsumer("late.0", "test", moduleInputChannel, rabbitConsumerProperties); properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload.equals('0') ? 0 : 1")); @@ -655,31 +657,31 @@ public class RabbitBinderTests extends PartitionCapableBinderTests partLateConsumerProperties = createConsumerProperties(); + partLateConsumerProperties.getExtension().setPrefix("latebinder."); partLateConsumerProperties.setPartitioned(true); partLateConsumerProperties.setInstanceIndex(0); Binding partlate0Consumer0Binding = binder.bindConsumer("partlate.0", "test", partInputChannel0, partLateConsumerProperties); partLateConsumerProperties.setInstanceIndex(1); Binding partlate0Consumer1Binding = binder.bindConsumer("partlate.0", "test", partInputChannel1, partLateConsumerProperties); - RabbitProducerProperties noDlqProducerProperties = new RabbitProducerProperties(); - noDlqProducerProperties.setPrefix("latebinder."); + ExtendedProducerProperties noDlqProducerProperties = createProducerProperties(); + noDlqProducerProperties.getExtension().setPrefix("latebinder."); MessageChannel noDLQOutputChannel = new DirectChannel(); Binding noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel, noDlqProducerProperties); QueueChannel noDLQInputChannel = new QueueChannel(); - RabbitConsumerProperties noDlqConsumerProperties = new RabbitConsumerProperties(); - noDlqConsumerProperties.setPrefix("latebinder."); + ExtendedConsumerProperties noDlqConsumerProperties = createConsumerProperties(); + noDlqConsumerProperties.getExtension().setPrefix("latebinder."); Binding noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel, noDlqConsumerProperties); MessageChannel outputChannel = new DirectChannel(); Binding pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel, noDlqProducerProperties); QueueChannel pubSubInputChannel = new QueueChannel(); - noDlqConsumerProperties.setDurableSubscription(false); + noDlqConsumerProperties.getExtension().setDurableSubscription(false); Binding nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup", pubSubInputChannel, noDlqConsumerProperties); QueueChannel durablePubSubInputChannel = new QueueChannel(); - noDlqConsumerProperties.setDurableSubscription(true); + noDlqConsumerProperties.getExtension().setDurableSubscription(true); Binding durableConsumerBinding = binder.bindConsumer("latePubSub", "lateDurableGroup", durablePubSubInputChannel, noDlqConsumerProperties); proxy.start(); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 81b40def5..76e56f328 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -23,6 +23,8 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.cloud.stream.binder.AbstractTestBinder; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.codec.kryo.PojoCodec; import org.springframework.integration.context.IntegrationContextUtils; @@ -37,7 +39,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; * @author David Turanski * @author Mark Fisher */ -public class RabbitTestBinder extends AbstractTestBinder { +public class RabbitTestBinder extends AbstractTestBinder, ExtendedProducerProperties> { private final RabbitAdmin rabbitAdmin; @@ -65,18 +67,20 @@ public class RabbitTestBinder extends AbstractTestBinder bindConsumer(String name, String group, MessageChannel moduleInputChannel, RabbitConsumerProperties properties) { + public Binding bindConsumer(String name, String group, MessageChannel moduleInputChannel, + ExtendedConsumerProperties properties) { if (group != null) { - this.queues.add(properties.getPrefix() + name + ("." + group)); + this.queues.add(properties.getExtension().getPrefix() + name + ("." + group)); } - this.exchanges.add(properties.getPrefix() + name); + this.exchanges.add(properties.getExtension().getPrefix() + name); return super.bindConsumer(name, group, moduleInputChannel, properties); } @Override - public Binding bindProducer(String name, MessageChannel moduleOutputChannel, RabbitProducerProperties properties) { - this.queues.add(properties.getPrefix() + name + ".default"); - this.exchanges.add(properties.getPrefix() + name); + public Binding bindProducer(String name, MessageChannel moduleOutputChannel, + ExtendedProducerProperties properties) { + this.queues.add(properties.getExtension().getPrefix() + name + ".default"); + this.exchanges.add(properties.getExtension().getPrefix() + name); return super.bindProducer(name, moduleOutputChannel, properties); } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index d7ea1c6fe..ab1ea2b56 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -23,6 +23,7 @@ import static org.hamcrest.CoreMatchers.not; import static org.hamcrest.collection.IsMapContaining.hasKey; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.List; @@ -36,6 +37,8 @@ import org.mockito.Mockito; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.boot.SpringApplication; import org.springframework.boot.actuate.health.CompositeHealthIndicator; @@ -45,11 +48,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binding.ChannelBindingService; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.messaging.MessageChannel; /** * @author Marius Bogoevici @@ -101,6 +107,45 @@ public class RabbitBinderModuleTests { assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP)); } + @Test + public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() { + context = SpringApplication.run(SimpleProcessor.class, + "--server.port=0", + "--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true", + "--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true"); + BinderFactory binderFactory = context.getBean(BinderFactory.class); + Binder binder = binderFactory.getBinder(null); + assertThat(binder, instanceOf(RabbitMessageChannelBinder.class)); + ChannelBindingService channelBindingService = context.getBean(ChannelBindingService.class); + DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(channelBindingService); + Map>> consumerBindings = (Map>>) + channelBindingServiceAccessor.getPropertyValue("consumerBindings"); + Binding inputBinding = consumerBindings.get("input").get(0); + SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding, + "endpoint.messageListenerContainer", + SimpleMessageListenerContainer.class); + assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class)); + Map> producerBindings = + (Map>) TestUtils.getPropertyValue(channelBindingService, "producerBindings"); + Binding outputBinding = producerBindings.get("output"); + assertTrue(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional", Boolean.class)); + DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); + ConnectionFactory binderConnectionFactory = + (ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory"); + assertThat(binderConnectionFactory, instanceOf(CachingConnectionFactory.class)); + ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); + assertThat(binderConnectionFactory, is(connectionFactory)); + CompositeHealthIndicator bindersHealthIndicator = + context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class); + DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator); + assertNotNull(bindersHealthIndicator); + @SuppressWarnings("unchecked") + Map healthIndicators = + (Map) directFieldAccessor.getPropertyValue("indicators"); + assertThat(healthIndicators, hasKey("rabbit")); + assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP)); + } + @Test public void testParentConnectionFactoryInheritedIfOverridden() { context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class).run("--server.port=0"); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedBindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedBindingProperties.java new file mode 100644 index 000000000..afb24d332 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedBindingProperties.java @@ -0,0 +1,30 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * Properties that extend the common binding properties for a particular binder implementation. + * + * @author Marius Bogoevici + * @author Mark Fisher + */ +public interface ExtendedBindingProperties { + + C getExtendedConsumerProperties(String channelName); + + P getExtendedProducerProperties(String channelName); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedConsumerProperties.java new file mode 100644 index 000000000..a203f4038 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedConsumerProperties.java @@ -0,0 +1,35 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * Extension of {@link ConsumerProperties} to be used with an {@link ExtendedPropertiesBinder}. + * + * @author Marius Bogoevici + */ +public class ExtendedConsumerProperties extends ConsumerProperties { + + private T extension; + + public ExtendedConsumerProperties(T extension) { + this.extension = extension; + } + + public T getExtension() { + return extension; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedProducerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedProducerProperties.java new file mode 100644 index 000000000..379b93310 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedProducerProperties.java @@ -0,0 +1,34 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * @author Marius Bogoevici + */ +public class ExtendedProducerProperties extends ProducerProperties { + + private T extension; + + public ExtendedProducerProperties(T extension) { + this.extension = extension; + } + + public T getExtension() { + return extension; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinder.java new file mode 100644 index 000000000..7c9d2835c --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinder.java @@ -0,0 +1,29 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * Extension of {@link Binder} that takes {@link ExtendedConsumerProperties} and {@link ExtendedProducerProperties} + * as arguments. In addition to supporting binding operations, it allows the binder to provide values for the + * additional properties it expects on the bindings. + * + * @author Marius Bogoevici + */ +public interface ExtendedPropertiesBinder + extends Binder, ExtendedProducerProperties

>, ExtendedBindingProperties { + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index a03080b8c..a4028b223 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java @@ -109,10 +109,7 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina @SuppressWarnings("unchecked") Binder binder = (Binder) binderFactory.getBinder(binderName); - Class producerPropertiesClass = - ChannelBindingService.resolveProducerPropertiesType(binder); - ProducerProperties producerProperties = - this.channelBindingServiceProperties.getProducerProperties(channelName, producerPropertiesClass); + ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(channelName); String destinationName = this.channelBindingServiceProperties.getBindingDestination(channelName); this.dynamicDestinationsBindable.addOutputBinding(beanName, binder.bindProducer(destinationName, channel, producerProperties)); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index 5015f5f10..79d99cc19 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -25,13 +25,16 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.BeanUtils; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; -import org.springframework.core.ResolvableType; import org.springframework.messaging.MessageChannel; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -71,12 +74,17 @@ public class ChannelBindingService { List> bindings = new ArrayList<>(); Binder binder = (Binder) getBinderForChannel(inputChannelName); - Class propertiesClass = resolveConsumerPropertiesType(binder); ConsumerProperties consumerProperties = - this.channelBindingServiceProperties.getConsumerProperties(inputChannelName, propertiesClass); + this.channelBindingServiceProperties.getConsumerProperties(inputChannelName); + if (binder instanceof ExtendedPropertiesBinder) { + ExtendedPropertiesBinder extendedPropertiesBinder = (ExtendedPropertiesBinder) binder; + Object extension = extendedPropertiesBinder.getExtendedConsumerProperties(inputChannelName); + ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(extension); + BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties); + consumerProperties = extendedConsumerProperties; + } for (String target : channelBindingTargets) { - Binding binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), - inputChannel, consumerProperties); + Binding binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), inputChannel, consumerProperties); bindings.add(binding); } this.consumerBindings.put(inputChannelName, bindings); @@ -88,9 +96,14 @@ public class ChannelBindingService { String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(outputChannelName); Binder binder = (Binder) getBinderForChannel(outputChannelName); - Class propertiesClass = resolveProducerPropertiesType(binder); - ProducerProperties producerProperties = - this.channelBindingServiceProperties.getProducerProperties(outputChannelName, propertiesClass); + ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(outputChannelName); + if (binder instanceof ExtendedPropertiesBinder) { + ExtendedPropertiesBinder extendedPropertiesBinder = (ExtendedPropertiesBinder) binder; + Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputChannelName); + ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(extension); + BeanUtils.copyProperties(producerProperties, extendedProducerProperties); + producerProperties = extendedProducerProperties; + } Binding binding = binder.bindProducer(channelBindingTarget, outputChannel, producerProperties); this.producerBindings.put(outputChannelName, binding); return binding; @@ -122,45 +135,4 @@ public class ChannelBindingService { String transport = this.channelBindingServiceProperties.getBinder(channelName); return binderFactory.getBinder(transport); } - - - static Class resolveConsumerPropertiesType(Binder binder) { - return resolveTypeForBinder(binder, ConsumerProperties.class); - } - - static Class resolveProducerPropertiesType(Binder binder) { - return resolveTypeForBinder(binder, ProducerProperties.class); - } - - @SuppressWarnings("unchecked") - static Class resolveTypeForBinder(Binder binder, Class upperBound) { - Class propertiesClass = null; - ResolvableType currentType = ResolvableType.forType(binder.getClass()); - while (!Object.class.equals(currentType.getRawClass()) && propertiesClass == null) { - ResolvableType[] interfaces = currentType.getInterfaces(); - ResolvableType binderResolvableType = null; - for (ResolvableType interfaceType : interfaces) { - if (Binder.class.equals(interfaceType.getRawClass())) { - binderResolvableType = interfaceType; - break; - } - } - if (binderResolvableType == null) { - currentType = currentType.getSuperType(); - } - else { - ResolvableType[] generics = binderResolvableType.getGenerics(); - for (ResolvableType generic : generics) { - Class resolvedParameter = generic.resolve(); - if (resolvedParameter != null && upperBound.isAssignableFrom(resolvedParameter)) { - propertiesClass = (Class) resolvedParameter; - } - } - } - } - if (propertiesClass == null) { - propertiesClass = upperBound; - } - return propertiesClass; - } } 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 0f3178c85..426660767 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 @@ -16,9 +16,15 @@ package org.springframework.cloud.stream.config; +import javax.validation.constraints.AssertTrue; + import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude.Include; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.validation.annotation.Validated; + /** * Contains the properties of a binding. * @author Marius Bogoevici @@ -26,6 +32,7 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include; * @author Gary Russell */ @JsonInclude(value = Include.NON_DEFAULT) +@Validated public class BindingProperties { private static final String COMMA = ","; @@ -49,6 +56,11 @@ public class BindingProperties { private String binder; + + private ConsumerProperties consumer = null; + + private ProducerProperties producer = null; + public String getDestination() { return this.destination; } @@ -81,6 +93,27 @@ public class BindingProperties { this.binder = binder; } + public ConsumerProperties getConsumer() { + return consumer; + } + + public void setConsumer(ConsumerProperties consumer) { + this.consumer = consumer; + } + + public ProducerProperties getProducer() { + return producer; + } + + public void setProducer(ProducerProperties producer) { + this.producer = producer; + } + + @AssertTrue(message = "A binding must not set both producer and consumer properties.") + public boolean onlyOneOfProducerOrConsumerSet() { + return consumer == null || producer == null; + } + public String toString() { StringBuilder sb = new StringBuilder(); sb.append("destination=" + this.destination); 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 f983920f8..4a5aa0bfd 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 @@ -16,10 +16,7 @@ package org.springframework.cloud.stream.config; -import java.beans.PropertyDescriptor; -import java.util.ArrayList; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.Properties; import java.util.TreeMap; @@ -27,14 +24,9 @@ import java.util.TreeMap; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude.Include; -import org.springframework.beans.BeanUtils; import org.springframework.beans.BeansException; -import org.springframework.beans.MutablePropertyValues; -import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.bind.PropertySourcesPropertyValues; -import org.springframework.boot.bind.RelaxedDataBinder; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; @@ -56,17 +48,6 @@ import org.springframework.util.StringUtils; @JsonInclude(Include.NON_DEFAULT) public class ChannelBindingServiceProperties implements ApplicationContextAware, InitializingBean { - private final static String[] bindingPropertyFields; - - static { - PropertyDescriptor[] propertyDescriptors = BeanUtils.getPropertyDescriptors(BindingProperties.class); - List propertyNames = new ArrayList<>(); - for (PropertyDescriptor propertyDescriptor : propertyDescriptors) { - propertyNames.add(propertyDescriptor.getName()); - } - bindingPropertyFields = propertyNames.toArray(new String[propertyNames.size()]); - } - private ConversionService conversionService; @Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}") @@ -197,48 +178,27 @@ public class ChannelBindingServiceProperties implements ApplicationContextAware, return properties; } - public T getConsumerProperties(String inputChannelName, Class beanClass) { + public ConsumerProperties getConsumerProperties(String inputChannelName) { Assert.notNull(inputChannelName, "The input channel name cannot be null"); - Assert.notNull(beanClass, "The bean class cannot be null"); - T consumerProperties = populateProperties(inputChannelName, beanClass, consumerDefaults); + ConsumerProperties consumerProperties = getBindingProperties(inputChannelName).getConsumer(); + if (consumerProperties == null) { + consumerProperties = new ConsumerProperties(); + } consumerProperties.setInstanceCount(this.instanceCount); consumerProperties.setInstanceIndex(this.instanceIndex); return consumerProperties; } - public T getProducerProperties(String outputChannelName, Class beanClass) { + public ProducerProperties getProducerProperties(String outputChannelName) { Assert.notNull(outputChannelName, "The output channel name cannot be null"); - Assert.notNull(beanClass, "The bean class cannot be null"); - T producerProperties = populateProperties(outputChannelName, beanClass, producerDefaults); + ProducerProperties producerProperties = getBindingProperties(outputChannelName).getProducer(); + if (producerProperties == null) { + producerProperties = new ProducerProperties(); + } return producerProperties; } - - private C populateProperties(String channelName, Class propertiesClass, Properties defaults) { - C beanInstance; - try { - beanInstance = propertiesClass.newInstance(); - } - catch (InstantiationException | IllegalAccessException e) { - throw new BeanInitializationException(e.getMessage()); - } - // bind defaults first - RelaxedDataBinder dataBinder = new RelaxedDataBinder(beanInstance); - dataBinder.setIgnoreUnknownFields(ignoreUnknownProperties); - dataBinder.setDisallowedFields(bindingPropertyFields); - dataBinder.bind(new MutablePropertyValues(defaults)); - // bind configured properties next, if available - if (applicationContext != null && applicationContext.getEnvironment() != null) { - dataBinder = new RelaxedDataBinder(beanInstance, "spring.cloud.stream.bindings." + channelName); - dataBinder.setConversionService(conversionService); - dataBinder.setIgnoreUnknownFields(ignoreUnknownProperties); - dataBinder.setDisallowedFields(bindingPropertyFields); - dataBinder.bind(new PropertySourcesPropertyValues(applicationContext.getEnvironment().getPropertySources())); - } - return beanInstance; - } - public BindingProperties getBindingProperties(String channelName) { BindingProperties bindingProperties = bindings.containsKey(channelName) ? bindings.get(channelName) : new BindingProperties(); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultSettingsTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultSettingsTests.java deleted file mode 100644 index cf9b73a2c..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultSettingsTests.java +++ /dev/null @@ -1,130 +0,0 @@ -/* - * Copyright 2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.collection.IsArrayContainingInOrder.arrayContaining; -import static org.mockito.Matchers.eq; -import static org.mockito.Matchers.same; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -import org.hamcrest.CoreMatchers; -import org.junit.Test; -import org.mockito.ArgumentCaptor; - -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Import; - -/** - * @author Marius Bogoevici - */ -public class DefaultSettingsTests { - - @SuppressWarnings("unchecked") - @Test - public void testDefaultSettings() { - ConfigurableApplicationContext applicationContext = createBuilder().run( - "--spring.cloud.stream.bindings.foo.destination=fooDest", - "--spring.cloud.stream.bindings.bar.destination=barDest", - "--spring.cloud.stream.bindings.baz.destination=bazDest", - "--spring.cloud.stream.bindings.qux.destination=quxDest", - "--spring.cloud.stream.bindings.foo.group=fooBarGroup", - "--spring.cloud.stream.bindings.bar.group=fooBarGroup", - "--spring.cloud.stream.consumerDefaults.concurrency=2", - "--spring.cloud.stream.producerDefaults.requiredGroups=quxbazReq1,quxbazReq2"); - - Binder binder = applicationContext.getBean(Binder.class); - FooChannels fooChannels = applicationContext.getBean(FooChannels.class); - - ArgumentCaptor fooConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class); - verify(binder).bindConsumer(eq("fooDest"), eq("fooBarGroup"), same(fooChannels.foo()), - fooConsumerProperties.capture()); - assertThat(fooConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(2)); - ArgumentCaptor barConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class); - verify(binder).bindConsumer(eq("barDest"), eq("fooBarGroup"), same(fooChannels.bar()), - barConsumerProperties.capture()); - assertThat(barConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(2)); - - ArgumentCaptor bazProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class); - verify(binder).bindProducer(eq("bazDest"), same(fooChannels.baz()), bazProducerProperties.capture()); - assertThat(bazProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq1", "quxbazReq2")); - - ArgumentCaptor quxProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class); - verify(binder).bindProducer(eq("quxDest"), same(fooChannels.qux()), quxProducerProperties.capture()); - assertThat(quxProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq1", "quxbazReq2")); - - verifyNoMoreInteractions(binder); - applicationContext.close(); - } - - @SuppressWarnings("unchecked") - @Test - public void testDefaultSettingsOverridden() { - ConfigurableApplicationContext applicationContext = createBuilder().run( - "--spring.cloud.stream.bindings.foo.destination=fooDest", - "--spring.cloud.stream.bindings.bar.destination=barDest", - "--spring.cloud.stream.bindings.baz.destination=bazDest", - "--spring.cloud.stream.bindings.qux.destination=quxDest", - "--spring.cloud.stream.bindings.foo.group=fooBarGroup", - "--spring.cloud.stream.bindings.bar.group=fooBarGroup", - "--spring.cloud.stream.consumerDefaults.concurrency=2", - "--spring.cloud.stream.bindings.bar.concurrency=4", - "--spring.cloud.stream.producerDefaults.requiredGroups=quxbazReq1,quxbazReq2", - "--spring.cloud.stream.bindings.qux.requiredGroups=quxbazReq3,quxbazReq4"); - - Binder binder = applicationContext.getBean(Binder.class); - FooChannels fooChannels = applicationContext.getBean(FooChannels.class); - - ArgumentCaptor fooConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class); - verify(binder).bindConsumer(eq("fooDest"), eq("fooBarGroup"), same(fooChannels.foo()), - fooConsumerProperties.capture()); - assertThat(fooConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(2)); - ArgumentCaptor barConsumerProperties = ArgumentCaptor.forClass(ConsumerProperties.class); - verify(binder).bindConsumer(eq("barDest"), eq("fooBarGroup"), same(fooChannels.bar()), - barConsumerProperties.capture()); - assertThat(barConsumerProperties.getValue().getConcurrency(), CoreMatchers.equalTo(4)); - - ArgumentCaptor bazProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class); - verify(binder).bindProducer(eq("bazDest"), same(fooChannels.baz()), bazProducerProperties.capture()); - assertThat(bazProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq1", "quxbazReq2")); - - ArgumentCaptor quxProducerProperties = ArgumentCaptor.forClass(ProducerProperties.class); - verify(binder).bindProducer(eq("quxDest"), same(fooChannels.qux()), quxProducerProperties.capture()); - assertThat(quxProducerProperties.getValue().getRequiredGroups(), arrayContaining("quxbazReq3", "quxbazReq4")); - - verifyNoMoreInteractions(binder); - applicationContext.close(); - } - - private SpringApplicationBuilder createBuilder() { - return new SpringApplicationBuilder(TestFooChannels.class) - .web(false); - } - - - @EnableBinding(FooChannels.class) - @EnableAutoConfiguration - @Import(MockBinderRegistryConfiguration.class) - public static class TestFooChannels { - - } -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/PropertiesClassResolutionTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/PropertiesClassResolutionTests.java deleted file mode 100644 index 9eef1a1e0..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/PropertiesClassResolutionTests.java +++ /dev/null @@ -1,161 +0,0 @@ -/* - * Copyright 2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binding; - -import org.junit.Assert; -import org.junit.Test; - -import org.springframework.cloud.stream.binder.AbstractBinder; -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.Binding; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ProducerProperties; - - -/** - * @author Marius Bogoevici - */ -public class PropertiesClassResolutionTests { - - @Test - public void testDefaultResolution() { - SimpleBinderImplementation testBinder = new SimpleBinderImplementation(); - Assert.assertEquals(ConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder)); - Assert.assertEquals(ProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder)); - } - - @Test - public void testBinderImplementorWithCustomTypes() { - BinderImplementationWithCustomTypes testBinder = new BinderImplementationWithCustomTypes(); - Assert.assertEquals(SubclassConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder)); - Assert.assertEquals(SubclassProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder)); - } - - @Test - public void testExtendsAbstractBinderWithDefaultTypes() { - ExtendsAbstractBinderWithDefaultTypes testBinder = new ExtendsAbstractBinderWithDefaultTypes(); - Assert.assertEquals(ConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder)); - Assert.assertEquals(ProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder)); - } - - @Test - public void testExtendsAbstractBinderWithCustomTypes() { - ExtendsAbstractBinderWithCustomTypes testBinder = new ExtendsAbstractBinderWithCustomTypes(); - Assert.assertEquals(SubclassConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder)); - Assert.assertEquals(SubclassProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder)); - } - - @Test - public void testGenericBinderWithDefaultTypes() { - GenericBinderWithDefaultTypes testBinder = new GenericBinderWithDefaultTypes(); - Assert.assertEquals(ConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder)); - Assert.assertEquals(ProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder)); - } - - @Test - public void testGenericBinderWithCustomTypes() { - GenericBinderWithCustomTypes testBinder = new GenericBinderWithCustomTypes(); - Assert.assertEquals(SubclassConsumerProperties.class, ChannelBindingService.resolveConsumerPropertiesType(testBinder)); - Assert.assertEquals(SubclassProducerProperties.class, ChannelBindingService.resolveProducerPropertiesType(testBinder)); - } - - private class SimpleBinderImplementation implements Binder { - - @Override - public Binding bindConsumer(String name, String group, Object inboundBindTarget, ConsumerProperties consumerProperties) { - return null; - } - - @Override - public Binding bindProducer(String name, Object outboundBindTarget, ProducerProperties producerProperties) { - return null; - } - } - - private class BinderImplementationWithCustomTypes implements Binder { - - @Override - public Binding bindConsumer(String name, String group, Object inboundBindTarget, SubclassConsumerProperties consumerProperties) { - return null; - } - - @Override - public Binding bindProducer(String name, Object outboundBindTarget, SubclassProducerProperties producerProperties) { - return null; - } - } - - private class ExtendsAbstractBinderWithDefaultTypes extends AbstractBinder { - - @Override - protected Binding doBindConsumer(String name, String group, Object inputTarget, ConsumerProperties properties) { - return null; - } - - @Override - protected Binding doBindProducer(String name, Object outboundBindTarget, ProducerProperties properties) { - return null; - } - } - - private class ExtendsAbstractBinderWithCustomTypes extends AbstractBinder { - - @Override - protected Binding doBindConsumer(String name, String group, Object inputTarget, SubclassConsumerProperties properties) { - return null; - } - - @Override - protected Binding doBindProducer(String name, Object outboundBindTarget, SubclassProducerProperties properties) { - return null; - } - } - - private class GenericBinderWithDefaultTypes extends AbstractBinder { - - @Override - protected Binding doBindConsumer(String name, String group, Object inputTarget, C properties) { - return null; - } - - @Override - protected Binding doBindProducer(String name, Object outboundBindTarget, P properties) { - return null; - } - } - - private class GenericBinderWithCustomTypes extends AbstractBinder { - - @Override - protected Binding doBindConsumer(String name, String group, Object inputTarget, C properties) { - return null; - } - - @Override - protected Binding doBindProducer(String name, Object outboundBindTarget, P properties) { - return null; - } - } - - private class SubclassConsumerProperties extends ConsumerProperties { - - } - - private class SubclassProducerProperties extends ProducerProperties { - - } -} diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties index c79a9ce74..3c200ee1a 100644 --- a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties @@ -1,5 +1,5 @@ spring.cloud.stream.bindings.input.destination=partIn -spring.cloud.stream.bindings.input.partitioned=true +spring.cloud.stream.bindings.input.consumer.partitioned=true spring.cloud.stream.instanceCount=2 spring.cloud.stream.instanceIndex=0 diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties index 87d462a3a..aee346650 100644 --- a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/partitioned-producer-test.properties @@ -1,4 +1,4 @@ spring.cloud.stream.bindings.output.destination=partOut -spring.cloud.stream.bindings.output.partitionKeyExpression=payload -spring.cloud.stream.bindings.output.partitionCount=3 +spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload +spring.cloud.stream.bindings.output.producer.partitionCount=3