Allow binders to not affect the defaulting process

Fixes #557

Allow for a `defaultCandidate` flag on a binder configuration that, when FALSE,
ensures that the binder does not interfere with the default configuration process.

spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -729,6 +729,9 @@ If your application should connect to more than one broker of the same type, you
 ====
 Turning on explicit binder configuration will disable the default binder configuration process altogether.
 If you do this, all binders in use must be included in the configuration.
+Frameworks that intend to use Spring Cloud Stream transparently may create binder configurations that can be referenced by name, but will not affect the default binder configuration.
+In order to do so, a binder configuration may have its the `defaultCandidate` flag set to false, e.g. `spring.cloud.stream.binders.<configurationName>.defaultCandidate=false`.
+This denotes a configuration that will exist independently of the default binder configuration process.
 ====
 
 For example, this is the typical configuration for a processor application which connects to two RabbitMQ broker instances:
@@ -760,6 +763,32 @@ spring:
                 host: <host2>
 ----
 
+=== Binder configuration properties
+
+The following properties are available when creating custom binder configurations.
+They must be prefixed with `spring.cloud.stream.binder.<configurationName>`.
+
+type::
+  The binder type.
+It typically references one of the binders found on the classpath, in particular a key in a `META-INF/spring.binders` file.
++
+By default, it has the same value as the configuration name.
+inheritEnvironment::
+  Whether the configuration will inherit the environment of the application itself.
++
+Default `true`.
+environment::
+  Root for a set of properties that can be used to customize the environment of the binder.
+When this is configured, the context in which the binder is being created is not a child of the application context.
+This allows for complete separation between the binder components and the application components.
++
+Default `empty`.
+defaultCandidate::
+  Whether the binder configuration is a candidate for being considered a default binder, or can be used only when explicitly referenced.
+This allows adding binder configurations without interfering with the default processing.
++
+Default `true`.
+
 
 === Implementation strategies
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java

@@ -34,15 +34,21 @@ public class BinderConfiguration {
 
 	private final boolean inheritEnvironment;
 
+	private final boolean defaultCandidate;
+
 	/**
 	 * @param binderType the binder type used by this configuration
 	 * @param properties the properties for setting up the binder
-	 * @param inheritEnvironment whether the binder should inherit the environment of the module
+	 * @param inheritEnvironment whether the binder should inherit the environment of the
+	 * module
+	 * @param defaultCandidate whether the binder is user defined
 	 */
-	public BinderConfiguration(BinderType binderType, Properties properties, boolean inheritEnvironment) {
+	public BinderConfiguration(BinderType binderType, Properties properties, boolean inheritEnvironment,
+			boolean defaultCandidate) {
 		this.binderType = binderType;
 		this.properties = properties;
 		this.inheritEnvironment = inheritEnvironment;
+		this.defaultCandidate = defaultCandidate;
 	}
 
 	public BinderType getBinderType() {
@@ -56,4 +62,8 @@ public class BinderConfiguration {
 	public boolean isInheritEnvironment() {
 		return inheritEnvironment;
 	}
+
+	public boolean isDefaultCandidate() {
+		return defaultCandidate;
+	}
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java

@@ -19,9 +19,11 @@ package org.springframework.cloud.stream.binder;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -95,8 +97,31 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 				throw new IllegalStateException(
 						"A default binder has been requested, but there there is no binder available");
 			}
-			else if (this.binderConfigurations.size() == 1) {
-				configurationName = this.binderConfigurations.keySet().iterator().next();
+			else if (!StringUtils.hasText(defaultBinder)) {
+				Set<String> defaultCandidateConfigurations = new HashSet<>();
+				for (Map.Entry<String, BinderConfiguration> binderConfigurationEntry : binderConfigurations
+						.entrySet()) {
+					if (binderConfigurationEntry.getValue().isDefaultCandidate()) {
+						defaultCandidateConfigurations.add(binderConfigurationEntry.getKey());
+					}
+				}
+				if (defaultCandidateConfigurations.size() == 1) {
+					this.defaultBinder = defaultCandidateConfigurations.iterator().next();
+					configurationName = this.defaultBinder;
+				}
+				else {
+					if (defaultCandidateConfigurations.size() > 1) {
+						throw new IllegalStateException(
+								"A default binder has been requested, but there is more than one binder available: "
+										+ StringUtils
+												.collectionToCommaDelimitedString(defaultCandidateConfigurations)
+										+ ", and" + " no default binder has been set.");
+					}
+					else {
+						throw new IllegalStateException(
+								"A default binder has been requested, but there there is no binder available");
+					}
+				}
 			}
 			else {
 				if (StringUtils.hasText(this.defaultBinder)) {
@@ -135,7 +160,8 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 			}
 			args.add("--spring.jmx.default-domain=" + defaultDomain + "binder." + configurationName);
 			List<Class<?>> configurationClasses =
-					new ArrayList<>(Arrays.asList(binderConfiguration.getBinderType().getConfigurationClasses()));
+					new ArrayList<Class<?>>(
+							Arrays.asList(binderConfiguration.getBinderType().getConfigurationClasses()));
 			SpringApplicationBuilder springApplicationBuilder =
 					new SpringApplicationBuilder()
 							.sources(configurationClasses.toArray(new Class<?>[]{}))
@@ -150,9 +176,11 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 				springApplicationBuilder.parent(context);
 			}
 			if (useApplicationContextAsParent || (environment != null && binderConfiguration.isInheritEnvironment())) {
-				StandardEnvironment binderEnvironment = new StandardEnvironment();
-				binderEnvironment.merge(environment);
-				springApplicationBuilder.environment(binderEnvironment);
+				if (environment != null) {
+					StandardEnvironment binderEnvironment = new StandardEnvironment();
+					binderEnvironment.merge(environment);
+					springApplicationBuilder.environment(binderEnvironment);
+				}
 			}
 			ConfigurableApplicationContext binderProducingContext =
 					springApplicationBuilder.run(args.toArray(new String[args.size()]));

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Enumeration;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Properties;
 
@@ -41,7 +42,6 @@ import org.springframework.core.io.UrlResource;
 import org.springframework.core.io.support.PropertiesLoaderUtils;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
-import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -55,30 +55,34 @@ public class BinderFactoryConfiguration {
 	public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry,
 			ChannelBindingServiceProperties channelBindingServiceProperties) {
 		Map<String, BinderConfiguration> binderConfigurations = new HashMap<>();
-		if (!CollectionUtils.isEmpty(channelBindingServiceProperties.getBinders())) {
-			for (Map.Entry<String, BinderProperties> binderEntry :
-					channelBindingServiceProperties.getBinders().entrySet()) {
-				BinderProperties binderProperties = binderEntry.getValue();
-				if (binderTypeRegistry.get(binderEntry.getKey()) != null) {
-					binderConfigurations.put(binderEntry.getKey(),
-							new BinderConfiguration(binderTypeRegistry.get(binderEntry.getKey()),
-									binderProperties.getEnvironment(), binderProperties.isInheritEnvironment()));
-				}
-				else {
-					Assert.hasText(binderProperties.getType(), "No 'type' property present for custom " +
-							"binder " + binderEntry.getKey());
-					BinderType binderType = binderTypeRegistry.get(binderProperties.getType());
-					Assert.notNull(binderType, "Binder type " + binderProperties.getType() + " is not defined");
-					binderConfigurations.put(binderEntry.getKey(),
-							new BinderConfiguration(binderType, binderProperties.getEnvironment(),
-									binderProperties.isInheritEnvironment()));
-				}
+		Map<String, BinderProperties> declaredBinders = channelBindingServiceProperties.getBinders();
+		boolean defaultCandidatesExist = false;
+		Iterator<Map.Entry<String, BinderProperties>> binderPropertiesIterator = declaredBinders.entrySet().iterator();
+		while (!defaultCandidatesExist && binderPropertiesIterator.hasNext()) {
+			defaultCandidatesExist = binderPropertiesIterator.next().getValue().isDefaultCandidate();
+		}
+		for (Map.Entry<String, BinderProperties> binderEntry : declaredBinders.entrySet()) {
+			BinderProperties binderProperties = binderEntry.getValue();
+			if (binderTypeRegistry.get(binderEntry.getKey()) != null) {
+				binderConfigurations.put(binderEntry.getKey(),
+						new BinderConfiguration(binderTypeRegistry.get(binderEntry.getKey()),
+								binderProperties.getEnvironment(), binderProperties.isInheritEnvironment(),
+								binderProperties.isDefaultCandidate()));
+			}
+			else {
+				Assert.hasText(binderProperties.getType(),
+						"No 'type' property present for custom " + "binder " + binderEntry.getKey());
+				BinderType binderType = binderTypeRegistry.get(binderProperties.getType());
+				Assert.notNull(binderType, "Binder type " + binderProperties.getType() + " is not defined");
+				binderConfigurations.put(binderEntry.getKey(),
+						new BinderConfiguration(binderType, binderProperties.getEnvironment(),
+								binderProperties.isInheritEnvironment(), binderProperties.isDefaultCandidate()));
 			}
 		}
-		else {
+		if (!defaultCandidatesExist) {
 			for (Map.Entry<String, BinderType> entry : binderTypeRegistry.getAll().entrySet()) {
 				binderConfigurations.put(entry.getKey(),
-						new BinderConfiguration(entry.getValue(), new Properties(), true));
+						new BinderConfiguration(entry.getValue(), new Properties(), true, true));
 			}
 		}
 		DefaultBinderFactory binderFactory = new DefaultBinderFactory<>(binderConfigurations);

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java

@@ -31,6 +31,8 @@ public class BinderProperties {
 
 	private boolean inheritEnvironment = true;
 
+	private boolean defaultCandidate = true;
+
 	public String getType() {
 		return type;
 	}
@@ -54,4 +56,12 @@ public class BinderProperties {
 	public void setInheritEnvironment(boolean inheritEnvironment) {
 		this.inheritEnvironment = inheritEnvironment;
 	}
+
+	public boolean isDefaultCandidate() {
+		return defaultCandidate;
+	}
+
+	public void setDefaultCandidate(boolean defaultCandidate) {
+		this.defaultCandidate = defaultCandidate;
+	}
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java

@@ -94,13 +94,15 @@ public class BinderFactoryConfigurationTests {
 	public void loadBinderTypeRegistryWithOneCustomBinderAndSharedEnvironment() throws Exception {
 		ConfigurableApplicationContext context = createBinderTestContext(
 				new String[] {"binder1"}, "binder1.name=foo",
-				"spring.cloud.stream.binders.custom.properties.foo=bar",
+				"spring.cloud.stream.binders.custom.environment.foo=bar",
 				"spring.cloud.stream.binders.custom.type=binder1");
 
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 
 		Binder binder1 = binderFactory.getBinder("custom");
 		assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo");
+
+		assertThat(binderFactory.getBinder(null)).isSameAs(binder1);
 	}
 
 	@Test
@@ -147,6 +149,38 @@ public class BinderFactoryConfigurationTests {
 		assertThat(binder2).isInstanceOf(StubBinder2.class);
 	}
 
+	@Test
+	public void loadBinderTypeRegistryWithCustomNonDefaultCandidate() throws Exception {
+
+		ConfigurableApplicationContext context = createBinderTestContext(
+				new String[] { "binder1"},
+				"spring.cloud.stream.binders.custom.type=binder1",
+				"spring.cloud.stream.binders.custom.environment.binder1.name=foo",
+				"spring.cloud.stream.binders.custom.defaultCandidate=false",
+				"spring.cloud.stream.binders.custom.inheritEnvironment=false");
+		BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class);
+		assertThat(binderTypeRegistry).isNotNull();
+		assertThat(binderTypeRegistry.getAll().size()).isEqualTo(1);
+		assertThat(binderTypeRegistry.getAll().keySet().contains("binder1"));
+		assertThat((Class[]) binderTypeRegistry.get("binder1").getConfigurationClasses())
+				.contains(StubBinder1Configuration.class);
+
+		BinderFactory binderFactory = context.getBean(BinderFactory.class);
+
+		Binder defaultBinder = binderFactory.getBinder(null);
+		assertThat(defaultBinder).isInstanceOf(StubBinder1.class);
+		assertThat(((StubBinder1) defaultBinder).getName()).isNullOrEmpty();
+
+		Binder binder1 = binderFactory.getBinder("binder1");
+		assertThat(binder1).isInstanceOf(StubBinder1.class);
+		assertThat(binder1).isSameAs(defaultBinder);
+
+		Binder custom = binderFactory.getBinder("custom");
+		assertThat(custom).isInstanceOf(StubBinder1.class);
+		assertThat(custom).isNotSameAs(defaultBinder);
+		assertThat(((StubBinder1) custom).getName()).isEqualTo("foo");
+	}
+
 	@Test
 	public void loadDefaultBinderWithTwoBinders() throws Exception {
 

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java

@@ -79,10 +79,9 @@ public class ChannelBindingServiceTests {
 		DefaultBinderFactory<MessageChannel> binderFactory =
 				new DefaultBinderFactory<>(Collections.singletonMap("mock",
 						new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}),
-								new Properties(), true)));
+								new Properties(), true, true)));
 		Binder binder = binderFactory.getBinder("mock");
-		ChannelBindingService service = new ChannelBindingService(properties,
-				binderFactory);
+		ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
 		MessageChannel inputChannel = new DirectChannel();
 		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
@@ -111,14 +110,9 @@ public class ChannelBindingServiceTests {
 
 		properties.setBindings(bindingProperties);
 
-		DefaultBinderFactory<MessageChannel> binderFactory = new DefaultBinderFactory<>(
-				Collections
-						.singletonMap("mock",
-								new BinderConfiguration(
-										new BinderType("mock",
-												new Class[] {
-														MockBinderConfiguration.class }),
-										new Properties(), true)));
+		DefaultBinderFactory<MessageChannel> binderFactory = new DefaultBinderFactory<>(Collections.singletonMap("mock",
+				new BinderConfiguration(new BinderType("mock", new Class[] { MockBinderConfiguration.class }),
+						new Properties(), true, true)));
 
 		Binder binder = binderFactory.getBinder("mock");
 		ChannelBindingService service = new ChannelBindingService(properties,
@@ -175,7 +169,7 @@ public class ChannelBindingServiceTests {
 										new BinderType("mock",
 												new Class[] {
 														MockBinderConfiguration.class }),
-										new Properties(), true)));
+										new Properties(), true, true)));
 		Binder binder = binderFactory.getBinder("mock");
 		ChannelBindingService service = new ChannelBindingService(properties,
 				binderFactory);
@@ -207,7 +201,7 @@ public class ChannelBindingServiceTests {
 										new BinderType("mock",
 												new Class[] {
 														MockBinderConfiguration.class }),
-										new Properties(), true)));
+										new Properties(), true, true)));
 		Binder binder = binderFactory.getBinder("mock");
 		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
@@ -277,7 +271,7 @@ public class ChannelBindingServiceTests {
 		DefaultBinderFactory<MessageChannel> binderFactory =
 				new DefaultBinderFactory<>(Collections.singletonMap("mock",
 						new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}),
-								new Properties(), true)));
+								new Properties(), true, true)));
 		ChannelBindingService service = new ChannelBindingService(serviceProperties, binderFactory);
 		MessageChannel outputChannel = new DirectChannel();
 		try {
@@ -301,14 +295,9 @@ public class ChannelBindingServiceTests {
 		final String inputChannelName = "input";
 		bindingProperties.put(inputChannelName, props);
 		serviceProperties.setBindings(bindingProperties);
-		DefaultBinderFactory<MessageChannel> binderFactory = new DefaultBinderFactory<>(
-				Collections
-						.singletonMap("mock",
-								new BinderConfiguration(
-										new BinderType("mock",
-												new Class[] {
-														MockBinderConfiguration.class }),
-										new Properties(), true)));
+		DefaultBinderFactory<MessageChannel> binderFactory = new DefaultBinderFactory<>(Collections.singletonMap("mock",
+				new BinderConfiguration(new BinderType("mock", new Class[] { MockBinderConfiguration.class }),
+						new Properties(), true, true)));
 		ChannelBindingService service = new ChannelBindingService(serviceProperties,
 				binderFactory);
 		MessageChannel inputChannel = new DirectChannel();