From ce2c26e9343785db3a0e10223d935a2057895e04 Mon Sep 17 00:00:00 2001
From: Brian Clozel <bclozel@pivotal.io>
Date: Wed, 18 Sep 2019 22:17:07 +0200
Subject: [PATCH] Use Reactor's new Schedulers.boundedElastic()

Prior to this commit, Spring Boot would use `Schedulers.elastic()` when
required to process blocking tasks in a reactive environment.
reactor/reactor-core#1804 introduced a new scheduler,
`Schedulers.boundedElastic()` that behaves quite similarly but:

* will limit the number of workers thread
* will queue tasks if no worker thread is available and reject them is
the queue is exceeds a limit

This allows Spring Boot to schedule blocking tasks as before and allows
greater flexibility.

Fixes gh-18269
See gh-18276
---
 .../web/reactive/AbstractWebFluxEndpointHandlerMapping.java  | 5 +++--
 .../boot/actuate/health/HealthIndicatorReactiveAdapter.java  | 2 +-
 .../boot/actuate/redis/RedisReactiveHealthIndicator.java     | 3 ++-
 spring-boot-project/spring-boot-dependencies/pom.xml         | 2 +-
 4 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java
index cddfa3b369..0744608d2c 100644
--- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java
+++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java
@@ -223,7 +223,8 @@ public abstract class AbstractWebFluxEndpointHandlerMapping extends RequestMappi
 
 	/**
 	 * An {@link OperationInvoker} that performs the invocation of a blocking operation on
-	 * a separate thread using Reactor's {@link Schedulers#elastic() elastic scheduler}.
+	 * a separate thread using Reactor's {@link Schedulers#boundedElastic() bounded
+	 * elastic scheduler}.
 	 */
 	protected static final class ElasticSchedulerInvoker implements OperationInvoker {
 
@@ -235,7 +236,7 @@ public abstract class AbstractWebFluxEndpointHandlerMapping extends RequestMappi
 
 		@Override
 		public Object invoke(InvocationContext context) {
-			return Mono.fromCallable(() -> this.invoker.invoke(context)).subscribeOn(Schedulers.elastic());
+			return Mono.fromCallable(() -> this.invoker.invoke(context)).subscribeOn(Schedulers.boundedElastic());
 		}
 
 	}
diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/health/HealthIndicatorReactiveAdapter.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/health/HealthIndicatorReactiveAdapter.java
index 9e29835ca1..1a6a3a6970 100644
--- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/health/HealthIndicatorReactiveAdapter.java
+++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/health/HealthIndicatorReactiveAdapter.java
@@ -43,7 +43,7 @@ public class HealthIndicatorReactiveAdapter implements ReactiveHealthIndicator {
 
 	@Override
 	public Mono<Health> health() {
-		return Mono.fromCallable(this.delegate::health).subscribeOn(Schedulers.elastic());
+		return Mono.fromCallable(this.delegate::health).subscribeOn(Schedulers.boundedElastic());
 	}
 
 }
diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicator.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicator.java
index 29f99c52af..7708fa8780 100644
--- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicator.java
+++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicator.java
@@ -56,7 +56,8 @@ public class RedisReactiveHealthIndicator extends AbstractReactiveHealthIndicato
 	}
 
 	private Mono<ReactiveRedisConnection> getConnection() {
-		return Mono.fromSupplier(this.connectionFactory::getReactiveConnection).subscribeOn(Schedulers.parallel());
+		return Mono.fromSupplier(this.connectionFactory::getReactiveConnection)
+				.subscribeOn(Schedulers.boundedElastic());
 	}
 
 	private Health up(Health.Builder builder, Properties info) {
diff --git a/spring-boot-project/spring-boot-dependencies/pom.xml b/spring-boot-project/spring-boot-dependencies/pom.xml
index 348b860a41..849128826e 100644
--- a/spring-boot-project/spring-boot-dependencies/pom.xml
+++ b/spring-boot-project/spring-boot-dependencies/pom.xml
@@ -171,7 +171,7 @@
 		<quartz.version>2.3.1</quartz.version>
 		<querydsl.version>4.2.1</querydsl.version>
 		<rabbit-amqp-client.version>5.7.3</rabbit-amqp-client.version>
-		<reactor-bom.version>Dysprosium-RC1</reactor-bom.version>
+		<reactor-bom.version>Dysprosium-BUILD-SNAPSHOT</reactor-bom.version>
 		<rest-assured.version>3.3.0</rest-assured.version>
 		<reactive-streams.version>1.0.3</reactive-streams.version>
 		<rsocket.version>1.0.0-RC3</rsocket.version>