Drop blocking RedisReactiveHealthIndicator calls
Update `RedisReactiveHealthIndicator` so that `getReactiveConnection` is not called directly since it blocks. Fixed gh-16756
This commit is contained in:
@@ -19,6 +19,7 @@ package org.springframework.boot.actuate.redis;
|
||||
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 @@ import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
|
||||
*
|
||||
* @author Stephane Nicoll
|
||||
* @author Mark Paluch
|
||||
* @author Artsiom Yudovin
|
||||
* @since 2.0.0
|
||||
*/
|
||||
public class RedisReactiveHealthIndicator extends AbstractReactiveHealthIndicator {
|
||||
@@ -44,10 +46,14 @@ public class RedisReactiveHealthIndicator extends AbstractReactiveHealthIndicato
|
||||
|
||||
@Override
|
||||
protected Mono<Health> doHealthCheck(Health.Builder builder) {
|
||||
ReactiveRedisConnection connection = this.connectionFactory
|
||||
.getReactiveConnection();
|
||||
return connection.serverCommands().info().map((info) -> up(builder, info))
|
||||
.doFinally((signal) -> connection.close());
|
||||
Mono<ReactiveRedisConnection> connection = Mono
|
||||
.fromSupplier(this.connectionFactory::getReactiveConnection)
|
||||
.subscribeOn(Schedulers.parallel());
|
||||
|
||||
return connection
|
||||
.flatMap((c) -> c.serverCommands().info().map((info) -> up(builder, info))
|
||||
.onErrorResume((e) -> Mono.just(builder.down(e).build()))
|
||||
.flatMap((signal) -> c.closeLater().thenReturn(signal)));
|
||||
}
|
||||
|
||||
private Health up(Health.Builder builder, Properties info) {
|
||||
|
||||
@@ -41,6 +41,7 @@ import static org.mockito.Mockito.verify;
|
||||
* @author Stephane Nicoll
|
||||
* @author Mark Paluch
|
||||
* @author Nikolay Rybak
|
||||
* @author Artsiom Yudovin
|
||||
*/
|
||||
public class RedisReactiveHealthIndicatorTests {
|
||||
|
||||
@@ -49,6 +50,7 @@ public class RedisReactiveHealthIndicatorTests {
|
||||
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 class RedisReactiveHealthIndicatorTests {
|
||||
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 class RedisReactiveHealthIndicatorTests {
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user