Extended default properties

Allow applications to configure default values for extended prducer and
consumer properties across multiple bindings in order to avoid repetition.

Requires https://github.com/spring-cloud/spring-cloud-stream/pull/1477

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";
 

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<RabbitConsumerProperties, RabbitProducerProperties> {
 
+	private static final String DEFAULTS_PREFIX = "spring.cloud.stream.rabbit.default";
+
 	private Map<String, RabbitBindingProperties> bindings = new HashMap<>();
 
 	public Map<String, RabbitBindingProperties> 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;
+	}
+
 }

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<ExtendedConsumerProperties<RabbitConsumerProperties>,
@@ -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<RabbitProducerProperties> producerProperties, MessageChannel errorChannel) {

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 {