Add health indicators infrastructure

- Adds a generic 'bindersHealthIndicator' bean controlled by `management.health.binders.enabled` that tallies results from binders
- Binders are expected to expose one or more health indicators based on the status of the middleware connection. If no health indicators are exposed, the status is deemed to be 'UNKNOWN'
- Add support for Redis and Rabbit based on existing health indicators

Moved binder health indicator to autoconfiguration

Also updated copyright

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -17,6 +17,9 @@
 package org.springframework.cloud.stream.binder.rabbit.config;
 
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.RabbitHealthIndicator;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
 import org.springframework.boot.autoconfigure.cloud.CloudAutoConfiguration;
@@ -66,5 +69,11 @@ public class RabbitServiceAutoConfiguration {
 	@Profile("!cloud")
 	@Import(RabbitAutoConfiguration.class)
 	protected static class NoCloudConfig {
+
+	}
+
+	@Bean
+	public HealthIndicator binderHealthIndicator(RabbitTemplate rabbitTemplate) {
+		return new RabbitHealthIndicator(rabbitTemplate);
 	}
 }

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java

@@ -16,13 +16,17 @@
 
 package org.springframework.cloud.stream.binder.rabbit.integration;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
 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 java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.After;
 import org.junit.ClassRule;
@@ -34,6 +38,9 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.actuate.health.CompositeHealthIndicator;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.Status;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.Binder;
@@ -73,7 +80,7 @@ public class RabbitBinderModuleTests {
 
 	@Test
 	public void testParentConnectionFactoryInheritedByDefault() {
-		context = SpringApplication.run(SimpleProcessor.class);
+		context = SpringApplication.run(SimpleProcessor.class, "--server.port=0");
 		BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
 		Binder<?> binder = binderFactory.getBinder(null);
 		assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
@@ -83,11 +90,20 @@ public class RabbitBinderModuleTests {
 		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<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) 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();
+		context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class).run("--server.port=0");
 		BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
 		Binder<?> binder = binderFactory.getBinder(null);
 		assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
@@ -97,6 +113,16 @@ public class RabbitBinderModuleTests {
 		assertThat(binderConnectionFactory, is(MOCK_CONNECTION_FACTORY));
 		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
 		assertThat(binderConnectionFactory, is(connectionFactory));
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		assertNotNull(bindersHealthIndicator);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, hasKey("rabbit"));
+		// mock connection factory behaves as if down
+		assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.DOWN));
 	}
 
 	@Test
@@ -106,6 +132,7 @@ public class RabbitBinderModuleTests {
 		params.add("--spring.cloud.stream.output.binder=custom");
 		params.add("--spring.cloud.stream.binders.custom.type=rabbit");
 		params.add("--spring.cloud.stream.binders.custom.environment.foo=bar");
+		params.add("--server.port=0");
 		context = SpringApplication.run(SimpleProcessor.class, params.toArray(new String[params.size()]));
 		BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
 		Binder<?> binder = binderFactory.getBinder(null);
@@ -115,6 +142,15 @@ public class RabbitBinderModuleTests {
 				(ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
 		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
 		assertThat(binderConnectionFactory, not(is(connectionFactory)));
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		assertNotNull(bindersHealthIndicator);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, hasKey("custom"));
+		assertThat(healthIndicators.get("custom").health().getStatus(), equalTo(Status.UP));
 	}
 
 	@EnableBinding(Processor.class)

spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder.redis.config;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
-import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.redis.RedisMessageChannelBinder;

spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisServiceAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.redis.config;
 
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.RedisHealthIndicator;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.cloud.CloudAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -66,4 +68,8 @@ public class RedisServiceAutoConfiguration {
 	protected static class NoCloudConfig {
 	}
 
+	@Bean
+	public HealthIndicator binderHealthIndicator(RedisConnectionFactory redisConnectionFactory) {
+		return new RedisHealthIndicator(redisConnectionFactory);
+	}
 }

spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/integration/RedisBinderModuleTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -16,13 +16,17 @@
 
 package org.springframework.cloud.stream.binder.redis.integration;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
 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 java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.After;
 import org.junit.ClassRule;
@@ -31,6 +35,9 @@ import org.mockito.Mockito;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.actuate.health.CompositeHealthIndicator;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.Status;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.Binder;
@@ -65,7 +72,7 @@ public class RedisBinderModuleTests {
 
 	@Test
 	public void testParentConnectionFactoryInheritedByDefault() {
-		context = SpringApplication.run(SimpleProcessor.class);
+		context = SpringApplication.run(SimpleProcessor.class, "--server.port=0");
 		BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
 		Binder<?> binder = binderFactory.getBinder(null);
 		assertThat(binder, instanceOf(RedisMessageChannelBinder.class));
@@ -75,6 +82,15 @@ public class RedisBinderModuleTests {
 		assertThat(binderConnectionFactory, instanceOf(RedisConnectionFactory.class));
 		RedisConnectionFactory connectionFactory = context.getBean(RedisConnectionFactory.class);
 		assertThat(binderConnectionFactory, is(connectionFactory));
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		assertNotNull(bindersHealthIndicator);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, hasKey("redis"));
+		assertThat(healthIndicators.get("redis").health().getStatus(), equalTo(Status.UP));
 	}
 
 	@Test
@@ -89,6 +105,15 @@ public class RedisBinderModuleTests {
 		assertThat(binderConnectionFactory, is(MOCK_CONNECTION_FACTORY));
 		RedisConnectionFactory connectionFactory = context.getBean(RedisConnectionFactory.class);
 		assertThat(binderConnectionFactory, is(connectionFactory));
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		assertNotNull(bindersHealthIndicator);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, hasKey("redis"));
+		assertThat(healthIndicators.get("redis").health().getStatus(), equalTo(Status.UP));
 	}
 
 	@Test
@@ -107,6 +132,15 @@ public class RedisBinderModuleTests {
 				(RedisConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
 		RedisConnectionFactory connectionFactory = context.getBean(RedisConnectionFactory.class);
 		assertThat(binderConnectionFactory, not(is(connectionFactory)));
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		assertNotNull(bindersHealthIndicator);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, hasKey("custom"));
+		assertThat(healthIndicators.get("custom").health().getStatus(), equalTo(Status.UP));
 	}
 
 	@EnableBinding(Processor.class)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -17,12 +17,20 @@
 package org.springframework.cloud.stream.binder;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.Banner.Mode;
+import org.springframework.boot.actuate.health.AbstractHealthIndicator;
+import org.springframework.boot.actuate.health.CompositeHealthIndicator;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.OrderedHealthAggregator;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -47,6 +55,8 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 
 	private volatile String defaultBinder;
 
+	private volatile CompositeHealthIndicator bindersHealthIndicator;
+
 	public DefaultBinderFactory(Map<String, BinderConfiguration> binderConfigurations) {
 		this.binderConfigurations = new HashMap<>(binderConfigurations);
 	}
@@ -57,6 +67,11 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 		this.context = (ConfigurableApplicationContext) applicationContext;
 	}
 
+	@Autowired(required = false)
+	public void setBindersHealthIndicator(CompositeHealthIndicator bindersHealthIndicator) {
+		this.bindersHealthIndicator = bindersHealthIndicator;
+	}
+
 	public void setDefaultBinder(String defaultBinder) {
 		this.defaultBinder = defaultBinder;
 	}
@@ -115,16 +130,19 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 				defaultDomain += ".";
 			}
 			args.add("--spring.jmx.default-domain=" + defaultDomain + "binder." + configurationName);
+			List<Class<?>> configurationClasses =
+					new ArrayList<>(Arrays.asList(binderConfiguration.getBinderType().getConfigurationClasses()));
 			SpringApplicationBuilder springApplicationBuilder =
 					new SpringApplicationBuilder()
-							.sources(binderConfiguration.getBinderType().getConfigurationClasses())
+							.sources(configurationClasses.toArray(new Class<?>[]{}))
 							.bannerMode(Mode.OFF)
 							.web(false);
 			// If the environment is not customized and a main context is available, we will set the latter as parent.
 			// This ensures that the defaults and user-defined customizations (e.g. custom connection factory beans)
 			// are propagated to the binder context. If the environment is customized, then the binder context should
 			// not inherit any beans from the parent
-			if (binderProperties.isEmpty() && context != null) {
+			boolean useApplicationContextAsParent = binderProperties.isEmpty() && context != null;
+			if (useApplicationContextAsParent) {
 				springApplicationBuilder.parent(context);
 			}
 			else if (environment != null && binderConfiguration.isInheritEnvironment()) {
@@ -136,6 +154,15 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 					springApplicationBuilder.run(args.toArray(new String[args.size()]));
 			@SuppressWarnings("unchecked")
 			Binder<T> binder = binderProducingContext.getBean(Binder.class);
+			if (bindersHealthIndicator != null) {
+				OrderedHealthAggregator healthAggregator = new OrderedHealthAggregator();
+				Map<String, HealthIndicator> indicators = binderProducingContext.getBeansOfType(HealthIndicator.class);
+				// if there are no health indicators in the child context, we just mark the binder's health as unknown
+				// this can happen due to the fact that configuration is inherited
+				HealthIndicator binderHealthIndicator =
+						indicators.isEmpty() ? new DefaultHealthIndicator() : new CompositeHealthIndicator(healthAggregator, indicators);
+				bindersHealthIndicator.addHealthIndicator(configurationName, binderHealthIndicator);
+			}
 			this.binderInstanceCache.put(configurationName, new BinderInstanceHolder<>(binder, binderProducingContext));
 		}
 		return this.binderInstanceCache.get(configurationName).getBinderInstance();
@@ -165,4 +192,12 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
 			return this.binderContext;
 		}
 	}
+
+	private static class DefaultHealthIndicator extends AbstractHealthIndicator {
+
+		@Override
+		protected void doHealthCheck(Health.Builder builder) throws Exception {
+			builder.unknown();
+		}
+	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -114,7 +114,6 @@ public class BinderFactoryConfiguration {
 		return new DefaultBinderTypeRegistry(binderTypes);
 	}
 
-
 	static Collection<BinderType> parseBinderConfigurations(ClassLoader classLoader, Resource resource)
 			throws IOException, ClassNotFoundException {
 		Properties properties = PropertiesLoaderUtils.loadProperties(resource);

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

@@ -19,6 +19,11 @@ package org.springframework.cloud.stream.config;
 import java.util.List;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.autoconfigure.ConditionalOnEnabledHealthIndicator;
+import org.springframework.boot.actuate.autoconfigure.EndpointAutoConfiguration;
+import org.springframework.boot.actuate.health.CompositeHealthIndicator;
+import org.springframework.boot.actuate.health.OrderedHealthAggregator;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -40,6 +45,7 @@ import org.springframework.messaging.MessageChannel;
 @Configuration
 @ConditionalOnBean(ChannelBindingService.class)
 @EnableConfigurationProperties(DefaultPollerProperties.class)
+@AutoConfigureBefore(EndpointAutoConfiguration.class)
 public class ChannelBindingAutoConfiguration {
 
 	@Autowired
@@ -59,4 +65,10 @@ public class ChannelBindingAutoConfiguration {
 		return new ChannelsEndpoint(adapters, properties);
 	}
 
+	@Bean
+	@ConditionalOnEnabledHealthIndicator("binders")
+	@ConditionalOnMissingBean(name = "bindersHealthIndicator")
+	public CompositeHealthIndicator bindersHealthIndicator() {
+		return new CompositeHealthIndicator(new OrderedHealthAggregator());
+	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -34,6 +34,7 @@ import java.net.URL;
 import java.net.URLClassLoader;
 
 import org.junit.Test;
+
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -186,8 +187,8 @@ public class BinderFactoryConfigurationTests {
 		Binder defaultBinder = binderFactory.getBinder(null);
 		assertThat(defaultBinder, is(binder2));
 	}
-
-	private static ConfigurableApplicationContext createBinderTestContext(String[] additionalClasspathDirectories,
+	
+	public static ConfigurableApplicationContext createBinderTestContext(String[] additionalClasspathDirectories,
 																		  String... properties)
 			throws IOException {
 		URL[] urls = ObjectUtils.isEmpty(additionalClasspathDirectories) ?

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

@@ -0,0 +1,98 @@
+/*
+ * 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.Matchers.instanceOf;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.Map;
+
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.collection.IsMapContaining;
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.boot.actuate.health.CompositeHealthIndicator;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.stub1.StubBinder1;
+import org.springframework.cloud.stream.binder.stub2.StubBinder2;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class HealthIndicatorsConfigurationTests {
+
+
+	@Test
+	public void healthIndicatorsCheck() throws Exception {
+		ConfigurableApplicationContext context =
+				createBinderTestContext(
+						new String[]{"binder1", "binder2"}, "spring.cloud.stream.defaultBinder:binder2");
+
+		Binder binder1 = context.getBean(BinderFactory.class).getBinder("binder1");
+		assertThat(binder1, instanceOf(StubBinder1.class));
+		Binder binder2 = context.getBean(BinderFactory.class).getBinder("binder2");
+		assertThat(binder2, instanceOf(StubBinder2.class));
+
+		CompositeHealthIndicator bindersHealthIndicator =
+				context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
+		DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
+		assertNotNull(bindersHealthIndicator);
+		@SuppressWarnings("unchecked")
+		Map<String,HealthIndicator> healthIndicators =
+				(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
+		assertThat(healthIndicators, IsMapContaining.hasKey("binder1"));
+		assertThat(healthIndicators.get("binder1").health().getStatus(), CoreMatchers.equalTo(Status.UP));
+		assertThat(healthIndicators, IsMapContaining.hasKey("binder2"));
+		assertThat(healthIndicators.get("binder2").health().getStatus(), CoreMatchers.equalTo(Status.UNKNOWN));
+	}
+
+	public static ConfigurableApplicationContext createBinderTestContext(String[] additionalClasspathDirectories,
+																		 String... properties)
+			throws IOException {
+		URL[] urls = ObjectUtils.isEmpty(additionalClasspathDirectories) ?
+				new URL[0] : new URL[additionalClasspathDirectories.length];
+		if (!ObjectUtils.isEmpty(additionalClasspathDirectories)) {
+			for (int i = 0; i < additionalClasspathDirectories.length; i++) {
+				urls[i] = new URL(new ClassPathResource(additionalClasspathDirectories[i]).getURL().toString() + "/");
+			}
+		}
+		ClassLoader classLoader = new URLClassLoader(urls, BinderFactoryConfigurationTests.class.getClassLoader());
+		return new SpringApplicationBuilder(SimpleSource.class)
+				.resourceLoader(new DefaultResourceLoader(classLoader))
+				.properties(properties)
+				.web(false)
+				.run();
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding
+	public static class SimpleSource {
+	}
+}

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

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.stub1;
 
+import org.springframework.boot.actuate.health.ApplicationHealthIndicator;
+import org.springframework.boot.actuate.health.HealthIndicator;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
@@ -34,4 +36,9 @@ public class StubBinder1Configuration {
 	public Binder<?> binder() {
 		return new StubBinder1();
 	}
+
+	@Bean
+	public HealthIndicator binderHealthIndicator() {
+		return new ApplicationHealthIndicator();
+	}
 }