Merge branch '2.1.x'

Closes gh-16860

Author: philwebb

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2017 the original author or authors.
+ * Copyright 2012-2019 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.
@@ -19,6 +19,7 @@
 import java.util.Properties;
 
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import org.springframework.boot.actuate.health.AbstractReactiveHealthIndicator;
 import org.springframework.boot.actuate.health.Health;
@@ -31,6 +32,7 @@
  *
  * @author Stephane Nicoll
  * @author Mark Paluch
+ * @author Artsiom Yudovin
  * @since 2.0.0
  */
 public class RedisReactiveHealthIndicator extends AbstractReactiveHealthIndicator {
@@ -44,15 +46,29 @@ public RedisReactiveHealthIndicator(
 
 	@Override
 	protected Mono<Health> doHealthCheck(Health.Builder builder) {
-		ReactiveRedisConnection connection = this.connectionFactory
-				.getReactiveConnection();
+		return getConnection()
+				.flatMap((connection) -> doHealthCheck(builder, connection));
+	}
+
+	private Mono<Health> doHealthCheck(Health.Builder builder,
+			ReactiveRedisConnection connection) {
 		return connection.serverCommands().info().map((info) -> up(builder, info))
-				.doFinally((signal) -> connection.close());
+				.onErrorResume((ex) -> Mono.just(down(builder, ex)))
+				.flatMap((health) -> connection.closeLater().thenReturn(health));
+	}
+
+	private Mono<ReactiveRedisConnection> getConnection() {
+		return Mono.fromSupplier(this.connectionFactory::getReactiveConnection)
+				.subscribeOn(Schedulers.parallel());
 	}
 
 	private Health up(Health.Builder builder, Properties info) {
 		return builder.up().withDetail(RedisHealthIndicator.VERSION,
 				info.getProperty(RedisHealthIndicator.REDIS_VERSION)).build();
 	}
 
+	private Health down(Health.Builder builder, Throwable cause) {
+		return builder.down(cause).build();
+	}
+
 }

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicatorTests.java

@@ -41,6 +41,7 @@
  * @author Stephane Nicoll
  * @author Mark Paluch
  * @author Nikolay Rybak
+ * @author Artsiom Yudovin
  */
 public class RedisReactiveHealthIndicatorTests {
 
@@ -49,6 +50,7 @@ public void redisIsUp() {
 		Properties info = new Properties();
 		info.put("redis_version", "2.8.9");
 		ReactiveRedisConnection redisConnection = mock(ReactiveRedisConnection.class);
+		given(redisConnection.closeLater()).willReturn(Mono.empty());
 		ReactiveServerCommands commands = mock(ReactiveServerCommands.class);
 		given(commands.info()).willReturn(Mono.just(info));
 		RedisReactiveHealthIndicator healthIndicator = createHealthIndicator(
@@ -59,7 +61,7 @@ public void redisIsUp() {
 			assertThat(h.getDetails()).containsOnlyKeys("version");
 			assertThat(h.getDetails().get("version")).isEqualTo("2.8.9");
 		}).verifyComplete();
-		verify(redisConnection).close();
+		verify(redisConnection).closeLater();
 	}
 
 	@Test
@@ -68,13 +70,14 @@ public void redisCommandIsDown() {
 		given(commands.info()).willReturn(
 				Mono.error(new RedisConnectionFailureException("Connection failed")));
 		ReactiveRedisConnection redisConnection = mock(ReactiveRedisConnection.class);
+		given(redisConnection.closeLater()).willReturn(Mono.empty());
 		RedisReactiveHealthIndicator healthIndicator = createHealthIndicator(
 				redisConnection, commands);
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 				.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.DOWN))
 				.verifyComplete();
-		verify(redisConnection).close();
+		verify(redisConnection).closeLater();
 	}
 
 	@Test