Optimize reactor operators in actuator support

This commit uses native Reactor operators `Mono.fromCallable` and
`subscribeOn` for better performance and support.

Fixes gh-17452

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java

@@ -26,7 +26,6 @@ import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoSink;
 import reactor.core.scheduler.Schedulers;
 
 import org.springframework.boot.actuate.endpoint.InvalidEndpointRequestException;
@@ -234,17 +233,7 @@ public abstract class AbstractWebFluxEndpointHandlerMapping extends RequestMappi
 
 		@Override
 		public Object invoke(InvocationContext context) {
-			return Mono.create((sink) -> Schedulers.elastic().schedule(() -> invoke(context, sink)));
-		}
-
-		private void invoke(InvocationContext context, MonoSink<Object> sink) {
-			try {
-				Object result = this.invoker.invoke(context);
-				sink.success(result);
-			}
-			catch (Exception ex) {
-				sink.error(ex);
-			}
+			return Mono.fromCallable(() -> this.invoker.invoke(context)).subscribeOn(Schedulers.elastic());
 		}
 
 	}

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/health/HealthIndicatorReactiveAdapter.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.
@@ -17,7 +17,6 @@
 package org.springframework.boot.actuate.health;
 
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoSink;
 import reactor.core.scheduler.Schedulers;
 
 import org.springframework.util.Assert;
@@ -40,17 +39,7 @@ public class HealthIndicatorReactiveAdapter implements ReactiveHealthIndicator {
 
 	@Override
 	public Mono<Health> health() {
-		return Mono.create((sink) -> Schedulers.elastic().schedule(() -> invoke(sink)));
-	}
-
-	private void invoke(MonoSink<Health> sink) {
-		try {
-			Health health = this.delegate.health();
-			sink.success(health);
-		}
-		catch (Exception ex) {
-			sink.error(ex);
-		}
+		return Mono.fromCallable(this.delegate::health).subscribeOn(Schedulers.elastic());
 	}
 
 }