diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java index a30709a5f..1d3a1bc28 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java @@ -19,13 +19,15 @@ package org.springframework.cloud.stream.binder.rabbit.properties; import org.hibernate.validator.constraints.Range; import org.springframework.amqp.core.ExchangeTypes; +import org.springframework.cloud.stream.config.MergableProperties; /** * @author Gary Russell + * @author Soby Chacko * @since 1.2 * */ -public abstract class RabbitCommonProperties { +public abstract class RabbitCommonProperties implements MergableProperties { public static final String DEAD_LETTER_EXCHANGE = "DLX"; diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java index 5172b4086..74f4e286c 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -26,10 +26,13 @@ import org.springframework.cloud.stream.binder.ExtendedBindingProperties; * @author Marius Bogoevici * @author Gary Russell * @author Oleg Zhurakousky + * @author Soby Chacko */ @ConfigurationProperties("spring.cloud.stream.rabbit") public class RabbitExtendedBindingProperties implements ExtendedBindingProperties { + private static final String DEFAULTS_PREFIX = "spring.cloud.stream.rabbit.default"; + private Map bindings = new HashMap<>(); public Map getBindings() { @@ -82,4 +85,14 @@ public class RabbitExtendedBindingProperties implements ExtendedBindingPropertie return properties; } + @Override + public String getDefaultsPrefix() { + return DEFAULTS_PREFIX; + } + + @Override + public Class getExtendedPropertiesEntryClass() { + return RabbitBindingProperties.class; + } + } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index a89115355..1c8d074f8 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -106,6 +106,7 @@ import com.rabbitmq.client.Envelope; * @author David Turanski * @author Marius Bogoevici * @author Artem Bilan + * @author Soby Chacko */ public class RabbitMessageChannelBinder extends AbstractMessageChannelBinder, @@ -236,6 +237,16 @@ public class RabbitMessageChannelBinder return this.extendedBindingProperties.getExtendedProducerProperties(channelName); } + @Override + public String getDefaultsPrefix() { + return this.extendedBindingProperties.getDefaultsPrefix(); + } + + @Override + public Class getExtendedPropertiesEntryClass() { + return this.extendedBindingProperties.getExtendedPropertiesEntryClass(); + } + @Override protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) { diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index c56591913..3fe4bf5a5 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -16,16 +16,12 @@ package org.springframework.cloud.stream.binder.rabbit.integration; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.UUID; - import org.junit.After; import org.junit.ClassRule; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; @@ -42,11 +38,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.Cloud; 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.ExtendedConsumerProperties; -import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.*; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; @@ -63,6 +55,11 @@ import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; + import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.mock; @@ -72,6 +69,7 @@ import static org.mockito.Mockito.verify; * @author Marius Bogoevici * @author Gary Russell * @author Artem Bilan + * @author Soby Chacko */ public class RabbitBinderModuleTests { @@ -287,6 +285,30 @@ public class RabbitBinderModuleTests { verify(cloud).getSingletonServiceConnector(ConnectionFactory.class, null); } + @Test + public void testExtendedProperties() { + context = new SpringApplicationBuilder(SimpleProcessor.class) + .web(WebApplicationType.NONE) + .run("--server.port=0", "--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey", + "--spring.cloud.stream.rabbit.bindings.output.producer.batch-size=512", + "--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4", + "--spring.cloud.stream.rabbit.bindings.input.consumer.exchange-type=fanout"); + BinderFactory binderFactory = context.getBean(BinderFactory.class); + Binder rabbitBinder = binderFactory.getBinder(null, MessageChannel.class); + + RabbitProducerProperties rabbitProducerProperties = + (RabbitProducerProperties)((ExtendedPropertiesBinder) rabbitBinder).getExtendedProducerProperties("output"); + + assertThat(rabbitProducerProperties.getRoutingKeyExpression()).isEqualTo("fooRoutingKey"); + assertThat(rabbitProducerProperties.getBatchSize()).isEqualTo(512); + + RabbitConsumerProperties rabbitConsumerProperties = + (RabbitConsumerProperties)((ExtendedPropertiesBinder) rabbitBinder).getExtendedConsumerProperties("input"); + + assertThat(rabbitConsumerProperties.getExchangeType()).isEqualTo(ExchangeTypes.FANOUT); + assertThat(rabbitConsumerProperties.getMaxConcurrency()).isEqualTo(4); + } + @EnableBinding(Processor.class) @SpringBootApplication public static class SimpleProcessor {