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

Extra information in WebFlux stacktraces

Browse Source 
Use the checkpoint operator at various places in WebFlux to insert
information that Reactor then uses to enrich exceptions, via suppressed
exceptions, when error signals flow through the operator.

Closes gh-22105
This commit is contained in:
Rossen Stoyanchev
2019-03-15 16:12:58 -04:00
parent 495ba2f58c
commit e6d206b45a
16 changed files with 144 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-webflux/src/main/java/org/springframework/web/reactive/DispatcherHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -67,11 +67,6 @@ import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
*/
public class DispatcherHandler implements WebHandler, ApplicationContextAware {
@SuppressWarnings("ThrowableInstanceNeverThrown")
private static final Exception HANDLER_NOT_FOUND_EXCEPTION =
new ResponseStatusException(HttpStatus.NOT_FOUND, "No matching handler");
@Nullable
private List<HandlerMapping> handlerMappings;
@@ -172,8 +167,13 @@ public class DispatcherHandler implements WebHandler, ApplicationContextAware {
private Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
return getResultHandler(result).handleResult(exchange, result)
.onErrorResume(ex -> result.applyExceptionHandler(ex).flatMap(exceptionResult ->
getResultHandler(exceptionResult).handleResult(exchange, exceptionResult)));
.checkpoint("Handler " + result.getHandler() + " [DispatcherHandler]")
.onErrorResume(ex ->
result.applyExceptionHandler(ex).flatMap(exResult -> {
String text = "Exception handler " + exResult.getHandler() +
", error=\"" + ex.getMessage() + "\" [DispatcherHandler]";
return getResultHandler(exResult).handleResult(exchange, exResult).checkpoint(text);
}));
}
private HandlerResultHandler getResultHandler(HandlerResult handlerResult) {

24
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultClientResponse.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -56,12 +56,17 @@ class DefaultClientResponse implements ClientResponse {
private final String logPrefix;
private final String requestDescription;
public DefaultClientResponse(ClientHttpResponse response, ExchangeStrategies strategies,
String logPrefix, String requestDescription) {
public DefaultClientResponse(ClientHttpResponse response, ExchangeStrategies strategies, String logPrefix) {
this.response = response;
this.strategies = strategies;
this.headers = new DefaultHeaders();
this.logPrefix = logPrefix;
this.requestDescription = requestDescription;
}
@@ -90,22 +95,35 @@ class DefaultClientResponse implements ClientResponse {
return this.response.getCookies();
}
@SuppressWarnings("unchecked")
@Override
public <T> T body(BodyExtractor<T, ? super ClientHttpResponse> extractor) {
return extractor.extract(this.response, new BodyExtractor.Context() {
T result = extractor.extract(this.response, new BodyExtractor.Context() {
@Override
public List<HttpMessageReader<?>> messageReaders() {
return strategies.messageReaders();
}
@Override
public Optional<ServerHttpResponse> serverResponse() {
return Optional.empty();
}
@Override
public Map<String, Object> hints() {
return Hints.from(Hints.LOG_PREFIX_HINT, logPrefix);
}
});
String description = "Body from " + this.requestDescription + " [DefaultClientResponse]";
if (result instanceof Mono) {
return (T) ((Mono<?>) result).checkpoint(description);
}
else if (result instanceof Flux) {
return (T) ((Flux<?>) result).checkpoint(description);
}
else {
return result;
}
}
@Override

4
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultClientResponseBuilder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -136,7 +136,7 @@ final class DefaultClientResponseBuilder implements ClientResponse.Builder {
// When building ClientResponse manually, the ClientRequest.logPrefix() has to be passed,
// e.g. via ClientResponse.Builder, but this (builder) is not used currently.
return new DefaultClientResponse(httpResponse, this.strategies, "");
return new DefaultClientResponse(httpResponse, this.strategies, "", "");
}

28
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java
View File

@@ -316,8 +316,9 @@ class DefaultWebClient implements WebClient {
ClientRequest request = (this.inserter != null ?
initRequestBuilder().body(this.inserter).build() :
initRequestBuilder().build());
return Mono.defer(() -> exchangeFunction.exchange(request))
.switchIfEmpty(NO_HTTP_CLIENT_RESPONSE_ERROR);
return Mono.defer(() -> exchangeFunction.exchange(request)
.checkpoint("Request to " + this.httpMethod.name() + " " + this.uri + " [DefaultWebClient]")
.switchIfEmpty(NO_HTTP_CLIENT_RESPONSE_ERROR));
}
private ClientRequest.Builder initRequestBuilder() {
@@ -445,8 +446,8 @@ class DefaultWebClient implements WebClient {
@Override
public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> elementType) {
return this.responseMono.flatMapMany(response -> handleBody(response,
response.bodyToFlux(elementType), mono -> mono.flatMapMany(Flux::error)));
return this.responseMono.flatMapMany(response ->
handleBody(response, response.bodyToFlux(elementType), mono -> mono.flatMapMany(Flux::error)));
}
private <T extends Publisher<?>> T handleBody(ClientResponse response,
@@ -459,7 +460,8 @@ class DefaultWebClient implements WebClient {
Mono<? extends Throwable> exMono = handler.apply(response, request);
exMono = exMono.flatMap(ex -> drainBody(response, ex));
exMono = exMono.onErrorResume(ex -> drainBody(response, ex));
return errorFunction.apply(exMono);
T result = errorFunction.apply(exMono);
return insertCheckpoint(result, response.statusCode(), request);
}
}
return bodyPublisher;
@@ -477,6 +479,22 @@ class DefaultWebClient implements WebClient {
.onErrorResume(ex2 -> Mono.empty()).thenReturn(ex);
}
@SuppressWarnings("unchecked")
private <T extends Publisher<?>> T insertCheckpoint(T result, HttpStatus status, HttpRequest request) {
String httpMethod = request.getMethodValue();
URI uri = request.getURI();
String description = status + " from " + httpMethod + " " + uri + " [DefaultWebClient]";
if (result instanceof Mono) {
return (T) ((Mono<?>) result).checkpoint(description);
}
else if (result instanceof Flux) {
return (T) ((Flux<?>) result).checkpoint(description);
}
else {
return result;
}
}
private static Mono<WebClientResponseException> createResponseException(
ClientResponse response, HttpRequest request) {

5
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ExchangeFunctions.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -105,7 +105,8 @@ public abstract class ExchangeFunctions {
.doOnCancel(() -> logger.debug(logPrefix + "Cancel signal (to close connection)"))
.map(httpResponse -> {
logResponse(httpResponse, logPrefix);
return new DefaultClientResponse(httpResponse, this.strategies, logPrefix);
return new DefaultClientResponse(
httpResponse, this.strategies, logPrefix, httpMethod.name() + " " + url);
});
}

9
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClientResponseException.java
View File

@@ -63,11 +63,16 @@ public class WebClientResponseException extends WebClientException {
* Constructor with response data only, and a default message.
* @since 5.1.4
*/
public WebClientResponseException(int statusCode, String statusText,
public WebClientResponseException(int status, String reasonPhrase,
@Nullable HttpHeaders headers, @Nullable byte[] body, @Nullable Charset charset,
@Nullable HttpRequest request) {
this(statusCode + " " + statusText, statusCode, statusText, headers, body, charset, request);
this(initMessage(status, reasonPhrase, request), status, reasonPhrase, headers, body, charset, request);
}
private static String initMessage(int status, String reasonPhrase, @Nullable HttpRequest request) {
return status + " " + reasonPhrase +
(request != null ? " from " + request.getMethodValue() + " " + request.getURI() : "");
}
/**

6
spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketHandlerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -74,7 +74,9 @@ public class JettyWebSocketHandlerAdapter {
@OnWebSocketConnect
public void onWebSocketConnect(Session session) {
this.delegateSession = this.sessionFactory.apply(session);
this.delegateHandler.handle(this.delegateSession).subscribe(this.delegateSession);
this.delegateHandler.handle(this.delegateSession)
.checkpoint(session.getUpgradeRequest().getRequestURI() + " [JettyWebSocketHandlerAdapter]")
.subscribe(this.delegateSession);
}
@OnWebSocketMessage

6
spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/StandardWebSocketHandlerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -80,7 +80,9 @@ public class StandardWebSocketHandlerAdapter extends Endpoint {
this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
});
this.delegateHandler.handle(this.delegateSession).subscribe(this.delegateSession);
this.delegateHandler.handle(this.delegateSession)
.checkpoint(session.getRequestURI() + " [StandardWebSocketHandlerAdapter]")
.subscribe(this.delegateSession);
}
private <T> WebSocketMessage toMessage(T message) {

2
spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java
View File

@@ -117,7 +117,7 @@ public class ReactorNettyWebSocketClient implements WebSocketClient {
if (logger.isDebugEnabled()) {
logger.debug("Started session '" + session.getId() + "' for " + url);
}
return handler.handle(session);
return handler.handle(session).checkpoint(url + " [ReactorNettyWebSocketClient]");
})
.doOnRequest(n -> {
if (logger.isDebugEnabled()) {

6
spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -204,7 +204,9 @@ public class UndertowWebSocketClient implements WebSocketClient {
channel.getReceiveSetter().set(adapter);
channel.resumeReceives();
handler.handle(session).subscribe(session);
handler.handle(session)
.checkpoint(url + " [UndertowWebSocketClient]")
.subscribe(session);
}
private HandshakeInfo createHandshakeInfo(URI url, DefaultNegotiation negotiation) {

6
spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/upgrade/ReactorNettyRequestUpgradeStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -16,6 +16,7 @@
package org.springframework.web.reactive.socket.server.upgrade;
import java.net.URI;
import java.util.function.Supplier;
import reactor.core.publisher.Mono;
@@ -81,7 +82,8 @@ public class ReactorNettyRequestUpgradeStrategy implements RequestUpgradeStrateg
ReactorNettyWebSocketSession session =
new ReactorNettyWebSocketSession(
in, out, handshakeInfo, bufferFactory, this.maxFramePayloadLength);
return handler.handle(session);
URI uri = exchange.getRequest().getURI();
return handler.handle(session).checkpoint(uri + " [ReactorNettyRequestUpgradeStrategy]");
});
}

8
spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/upgrade/UndertowRequestUpgradeStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-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.
@@ -93,14 +93,16 @@ public class UndertowRequestUpgradeStrategy implements RequestUpgradeStrategy {
}
@Override
public void onConnect(WebSocketHttpExchange httpExchange, WebSocketChannel channel) {
public void onConnect(WebSocketHttpExchange exchange, WebSocketChannel channel) {
UndertowWebSocketSession session = createSession(channel);
UndertowWebSocketHandlerAdapter adapter = new UndertowWebSocketHandlerAdapter(session);
channel.getReceiveSetter().set(adapter);
channel.resumeReceives();
this.handler.handle(session).subscribe(session);
this.handler.handle(session)
.checkpoint(exchange.getRequestURI() + " [UndertowRequestUpgradeStrategy]")
.subscribe(session);
}
private UndertowWebSocketSession createSession(WebSocketChannel channel) {

2
spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultClientResponseTests.java
View File

@@ -70,7 +70,7 @@ public class DefaultClientResponseTests {
public void createMocks() {
mockResponse = mock(ClientHttpResponse.class);
mockExchangeStrategies = mock(ExchangeStrategies.class);
defaultClientResponse = new DefaultClientResponse(mockResponse, mockExchangeStrategies, "");
defaultClientResponse = new DefaultClientResponse(mockResponse, mockExchangeStrategies, "", "");
}