Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add CF support for reactive actuators

Browse Source 
Closes gh-10780
This commit is contained in:
Madhura Bhave
2017-11-02 18:00:55 -07:00
parent 7c5d2fadd3
commit d50fe8874f
32 changed files with 2474 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java Normal file
View File

@@ -0,0 +1,196 @@
/*
* Copyright 2012-2017 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.boot.actuate.endpoint.web.reactive;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Map;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Schedulers;
import org.springframework.boot.actuate.endpoint.EndpointInfo;
import org.springframework.boot.actuate.endpoint.OperationInvoker;
import org.springframework.boot.actuate.endpoint.web.EndpointMediaTypes;
import org.springframework.boot.actuate.endpoint.web.OperationRequestPredicate;
import org.springframework.boot.actuate.endpoint.web.WebEndpointOperation;
import org.springframework.boot.endpoint.web.EndpointMapping;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.reactive.result.condition.ConsumesRequestCondition;
import org.springframework.web.reactive.result.condition.PatternsRequestCondition;
import org.springframework.web.reactive.result.condition.ProducesRequestCondition;
import org.springframework.web.reactive.result.condition.RequestMethodsRequestCondition;
import org.springframework.web.reactive.result.method.RequestMappingInfo;
import org.springframework.web.reactive.result.method.RequestMappingInfoHandlerMapping;
import org.springframework.web.util.pattern.PathPatternParser;
/**
* A custom {@link HandlerMapping} that makes web endpoints available over HTTP using
* Spring WebFlux.
*
* @author Andy Wilkinson
* @author Madhura Bhave
*/
public abstract class AbstractWebFluxEndpointHandlerMapping extends RequestMappingInfoHandlerMapping {
private static final PathPatternParser pathPatternParser = new PathPatternParser();
private final EndpointMapping endpointMapping;
private final Collection<EndpointInfo<WebEndpointOperation>> webEndpoints;
private final EndpointMediaTypes endpointMediaTypes;
private final CorsConfiguration corsConfiguration;
/**
* Creates a new {@code WebEndpointHandlerMapping} that provides mappings for the
* operations of the given {@code webEndpoints}.
* @param endpointMapping the base mapping for all endpoints
* @param collection the web endpoints
* @param endpointMediaTypes media types consumed and produced by the endpoints
*/
public AbstractWebFluxEndpointHandlerMapping(EndpointMapping endpointMapping,
Collection<EndpointInfo<WebEndpointOperation>> collection,
EndpointMediaTypes endpointMediaTypes) {
this(endpointMapping, collection, endpointMediaTypes, null);
}
/**
* Creates a new {@code WebEndpointHandlerMapping} that provides mappings for the
* operations of the given {@code webEndpoints}.
* @param endpointMapping the base mapping for all endpoints
* @param webEndpoints the web endpoints
* @param endpointMediaTypes media types consumed and produced by the endpoints
* @param corsConfiguration the CORS configuration for the endpoints
*/
public AbstractWebFluxEndpointHandlerMapping(EndpointMapping endpointMapping,
Collection<EndpointInfo<WebEndpointOperation>> webEndpoints,
EndpointMediaTypes endpointMediaTypes, CorsConfiguration corsConfiguration) {
this.endpointMapping = endpointMapping;
this.webEndpoints = webEndpoints;
this.endpointMediaTypes = endpointMediaTypes;
this.corsConfiguration = corsConfiguration;
setOrder(-100);
}
@Override
protected void initHandlerMethods() {
this.webEndpoints.stream()
.flatMap((webEndpoint) -> webEndpoint.getOperations().stream())
.forEach(this::registerMappingForOperation);
if (StringUtils.hasText(this.endpointMapping.getPath())) {
registerLinksMapping();
}
}
private void registerLinksMapping() {
registerMapping(
new RequestMappingInfo(
new PatternsRequestCondition(
pathPatternParser.parse(this.endpointMapping.getPath())),
new RequestMethodsRequestCondition(RequestMethod.GET), null, null,
null,
new ProducesRequestCondition(
this.endpointMediaTypes.getProduced()
.toArray(new String[this.endpointMediaTypes
.getProduced().size()])),
null),
this, getLinks());
}
protected RequestMappingInfo createRequestMappingInfo(
WebEndpointOperation operationInfo) {
OperationRequestPredicate requestPredicate = operationInfo.getRequestPredicate();
PatternsRequestCondition patterns = new PatternsRequestCondition(pathPatternParser
.parse(this.endpointMapping.createSubPath(requestPredicate.getPath())));
RequestMethodsRequestCondition methods = new RequestMethodsRequestCondition(
RequestMethod.valueOf(requestPredicate.getHttpMethod().name()));
ConsumesRequestCondition consumes = new ConsumesRequestCondition(
toStringArray(requestPredicate.getConsumes()));
ProducesRequestCondition produces = new ProducesRequestCondition(
toStringArray(requestPredicate.getProduces()));
return new RequestMappingInfo(null, patterns, methods, null, null, consumes,
produces, null);
}
private String[] toStringArray(Collection<String> collection) {
return collection.toArray(new String[collection.size()]);
}
@Override
protected CorsConfiguration initCorsConfiguration(Object handler, Method method,
RequestMappingInfo mapping) {
return this.corsConfiguration;
}
public Collection<EndpointInfo<WebEndpointOperation>> getEndpoints() {
return this.webEndpoints;
}
protected abstract Method getLinks();
protected abstract void registerMappingForOperation(WebEndpointOperation operation);
@Override
protected boolean isHandler(Class<?> beanType) {
return false;
}
@Override
protected RequestMappingInfo getMappingForMethod(Method method,
Class<?> handlerType) {
return null;
}
/**
* An {@link OperationInvoker} that performs the invocation of a blocking operation on
* a separate thread using Reactor's {@link Schedulers#elastic() elastic scheduler}.
*/
protected static final class ElasticSchedulerOperationInvoker
implements OperationInvoker {
private final OperationInvoker delegate;
public ElasticSchedulerOperationInvoker(OperationInvoker delegate) {
this.delegate = delegate;
}
@Override
public Object invoke(Map<String, Object> arguments) {
return Mono.create((sink) -> Schedulers.elastic()
.schedule(() -> invoke(arguments, sink)));
}
private void invoke(Map<String, Object> arguments, MonoSink<Object> sink) {
try {
Object result = this.delegate.invoke(arguments);
sink.success(result);
}
catch (Exception ex) {
sink.error(ex);
}
}
}
}

129
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/WebFluxEndpointHandlerMapping.java
View File

@@ -24,8 +24,6 @@ import java.util.Map;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Schedulers;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.actuate.endpoint.EndpointInfo;
@@ -36,7 +34,6 @@ import org.springframework.boot.actuate.endpoint.ParametersMissingException;
import org.springframework.boot.actuate.endpoint.web.EndpointLinksResolver;
import org.springframework.boot.actuate.endpoint.web.EndpointMediaTypes;
import org.springframework.boot.actuate.endpoint.web.Link;
import org.springframework.boot.actuate.endpoint.web.OperationRequestPredicate;
import org.springframework.boot.actuate.endpoint.web.WebEndpointOperation;
import org.springframework.boot.actuate.endpoint.web.WebEndpointResponse;
import org.springframework.boot.endpoint.web.EndpointMapping;
@@ -45,21 +42,12 @@ import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.reactive.result.condition.ConsumesRequestCondition;
import org.springframework.web.reactive.result.condition.PatternsRequestCondition;
import org.springframework.web.reactive.result.condition.ProducesRequestCondition;
import org.springframework.web.reactive.result.condition.RequestMethodsRequestCondition;
import org.springframework.web.reactive.result.method.RequestMappingInfo;
import org.springframework.web.reactive.result.method.RequestMappingInfoHandlerMapping;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.util.UriComponentsBuilder;
import org.springframework.web.util.pattern.PathPatternParser;
/**
* A custom {@link HandlerMapping} that makes web endpoints available over HTTP using
@@ -68,10 +56,7 @@ import org.springframework.web.util.pattern.PathPatternParser;
* @author Andy Wilkinson
* @since 2.0.0
*/
public class WebFluxEndpointHandlerMapping extends RequestMappingInfoHandlerMapping
implements InitializingBean {
private static final PathPatternParser pathPatternParser = new PathPatternParser();
public class WebFluxEndpointHandlerMapping extends AbstractWebFluxEndpointHandlerMapping implements InitializingBean {
private final Method handleRead = ReflectionUtils
.findMethod(ReadOperationHandler.class, "handle", ServerWebExchange.class);
@@ -84,14 +69,6 @@ public class WebFluxEndpointHandlerMapping extends RequestMappingInfoHandlerMapp
private final EndpointLinksResolver endpointLinksResolver = new EndpointLinksResolver();
private final EndpointMapping endpointMapping;
private final Collection<EndpointInfo<WebEndpointOperation>> webEndpoints;
private final EndpointMediaTypes endpointMediaTypes;
private final CorsConfiguration corsConfiguration;
/**
* Creates a new {@code WebEndpointHandlerMapping} that provides mappings for the
* operations of the given {@code webEndpoints}.
@@ -116,45 +93,17 @@ public class WebFluxEndpointHandlerMapping extends RequestMappingInfoHandlerMapp
public WebFluxEndpointHandlerMapping(EndpointMapping endpointMapping,
Collection<EndpointInfo<WebEndpointOperation>> webEndpoints,
EndpointMediaTypes endpointMediaTypes, CorsConfiguration corsConfiguration) {
this.endpointMapping = endpointMapping;
this.webEndpoints = webEndpoints;
this.endpointMediaTypes = endpointMediaTypes;
this.corsConfiguration = corsConfiguration;
super(endpointMapping, webEndpoints, endpointMediaTypes, corsConfiguration);
setOrder(-100);
}
@Override
protected void initHandlerMethods() {
this.webEndpoints.stream()
.flatMap((webEndpoint) -> webEndpoint.getOperations().stream())
.forEach(this::registerMappingForOperation);
if (StringUtils.hasText(this.endpointMapping.getPath())) {
registerLinksMapping();
}
}
private void registerLinksMapping() {
registerMapping(
new RequestMappingInfo(
new PatternsRequestCondition(
pathPatternParser.parse(this.endpointMapping.getPath())),
new RequestMethodsRequestCondition(RequestMethod.GET), null, null,
null,
new ProducesRequestCondition(
this.endpointMediaTypes.getProduced()
.toArray(new String[this.endpointMediaTypes
.getProduced().size()])),
null),
this, this.links);
protected Method getLinks() {
return this.links;
}
@Override
protected CorsConfiguration initCorsConfiguration(Object handler, Method method,
RequestMappingInfo mapping) {
return this.corsConfiguration;
}
private void registerMappingForOperation(WebEndpointOperation operation) {
protected void registerMappingForOperation(WebEndpointOperation operation) {
OperationType operationType = operation.getType();
OperationInvoker operationInvoker = operation.getInvoker();
if (operation.isBlocking()) {
@@ -162,50 +111,19 @@ public class WebFluxEndpointHandlerMapping extends RequestMappingInfoHandlerMapp
}
registerMapping(createRequestMappingInfo(operation),
operationType == OperationType.WRITE
? new WriteOperationHandler(operationInvoker)
: new ReadOperationHandler(operationInvoker),
? new WebFluxEndpointHandlerMapping.WriteOperationHandler(operationInvoker)
: new WebFluxEndpointHandlerMapping.ReadOperationHandler(operationInvoker),
operationType == OperationType.WRITE ? this.handleWrite
: this.handleRead);
}
private RequestMappingInfo createRequestMappingInfo(
WebEndpointOperation operationInfo) {
OperationRequestPredicate requestPredicate = operationInfo.getRequestPredicate();
PatternsRequestCondition patterns = new PatternsRequestCondition(pathPatternParser
.parse(this.endpointMapping.createSubPath(requestPredicate.getPath())));
RequestMethodsRequestCondition methods = new RequestMethodsRequestCondition(
RequestMethod.valueOf(requestPredicate.getHttpMethod().name()));
ConsumesRequestCondition consumes = new ConsumesRequestCondition(
toStringArray(requestPredicate.getConsumes()));
ProducesRequestCondition produces = new ProducesRequestCondition(
toStringArray(requestPredicate.getProduces()));
return new RequestMappingInfo(null, patterns, methods, null, null, consumes,
produces, null);
}
private String[] toStringArray(Collection<String> collection) {
return collection.toArray(new String[collection.size()]);
}
@Override
protected boolean isHandler(Class<?> beanType) {
return false;
}
@Override
protected RequestMappingInfo getMappingForMethod(Method method,
Class<?> handlerType) {
return null;
}
@ResponseBody
private Map<String, Map<String, Link>> links(ServerHttpRequest request) {
return Collections.singletonMap("_links",
this.endpointLinksResolver.resolveLinks(this.webEndpoints,
this.endpointLinksResolver.resolveLinks(getEndpoints(),
UriComponentsBuilder.fromUri(request.getURI()).replaceQuery(null)
.toUriString()));
}
/**
* Base class for handlers for endpoint operations.
*/
@@ -286,35 +204,4 @@ public class WebFluxEndpointHandlerMapping extends RequestMappingInfoHandlerMapp
}
/**
* An {@link OperationInvoker} that performs the invocation of a blocking operation on
* a separate thread using Reactor's {@link Schedulers#elastic() elastic scheduler}.
*/
private static final class ElasticSchedulerOperationInvoker
implements OperationInvoker {
private final OperationInvoker delegate;
private ElasticSchedulerOperationInvoker(OperationInvoker delegate) {
this.delegate = delegate;
}
@Override
public Object invoke(Map<String, Object> arguments) {
return Mono.create((sink) -> Schedulers.elastic()
.schedule(() -> invoke(arguments, sink)));
}
private void invoke(Map<String, Object> arguments, MonoSink<Object> sink) {
try {
Object result = this.delegate.invoke(arguments);
sink.success(result);
}
catch (Exception ex) {
sink.error(ex);
}
}
}
}