Cwikius-Spring/spring-data-elasticsearch
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-data-elasticsearch.git synced 2025-06-22 12:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

DATAES-863 - Improve server error response handling.

Browse Source 
Original PR: #480

(cherry picked from commit 3c44a1c96996ff2af496500505a8194e22b3de02)
This commit is contained in:
Peter-Josef Meisch 2020-06-11 19:13:59 +02:00
parent 4d2e4ac22c
commit 73fc8f65ee
1 changed files with 60 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java
View File

@ -804,53 +804,82 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
private <T> Publisher<? extends T> handleServerError(Request request, ClientResponse response) { private <T> Publisher<? extends T> handleServerError(Request request, ClientResponse response) {
RestStatus status = RestStatus.fromCode(response.statusCode().value()); int statusCode = response.statusCode().value();
RestStatus status = RestStatus.fromCode(statusCode);
String mediaType = response.headers().contentType().map(MediaType::toString).orElse(XContentType.JSON.mediaType());
return Mono.error(new ElasticsearchStatusException(String.format("%s request to %s returned error code %s.", return response.body(BodyExtractors.toMono(byte[].class)) //
request.getMethod(), request.getEndpoint(), response.statusCode().value()), status)); .map(bytes -> new String(bytes, StandardCharsets.UTF_8)) //
.flatMap(content -> contentOrError(content, mediaType, status))
.flatMap(unused -> Mono
.error(new ElasticsearchStatusException(String.format("%s request to %s returned error code %s.",
request.getMethod(), request.getEndpoint(), statusCode), status)));
} }
private <T> Publisher<? extends T> handleClientError(String logId, Request request, ClientResponse response, private <T> Publisher<? extends T> handleClientError(String logId, Request request, ClientResponse response,
Class<T> responseType) { Class<T> responseType) {
int statusCode = response.statusCode().value();
RestStatus status = RestStatus.fromCode(statusCode);
String mediaType = response.headers().contentType().map(MediaType::toString).orElse(XContentType.JSON.mediaType());
return response.body(BodyExtractors.toMono(byte[].class)) // return response.body(BodyExtractors.toMono(byte[].class)) //
.map(bytes -> new String(bytes, StandardCharsets.UTF_8)) // .map(bytes -> new String(bytes, StandardCharsets.UTF_8)) //
.flatMap(content -> { .flatMap(content -> contentOrError(content, mediaType, status)) //
String mediaType = response.headers().contentType().map(MediaType::toString) .doOnNext(content -> ClientLogger.logResponse(logId, response.statusCode(), content)) //
.orElse(XContentType.JSON.mediaType());
RestStatus status = RestStatus.fromCode(response.statusCode().value());
try {
ElasticsearchException exception = getElasticsearchException(response, content, mediaType);
if (exception != null) {
StringBuilder sb = new StringBuilder();
buildExceptionMessages(sb, exception);
return Mono.error(new ElasticsearchStatusException(sb.toString(), status, exception));
}
} catch (Exception e) {
return Mono.error(new ElasticsearchStatusException(content, status));
}
return Mono.just(content);
}).doOnNext(it -> ClientLogger.logResponse(logId, response.statusCode(), it)) //
.flatMap(content -> doDecode(response, responseType, content)); .flatMap(content -> doDecode(response, responseType, content));
} }
// region ElasticsearchException helper // region ElasticsearchException helper
/**
* checks if the given content body contains an {@link ElasticsearchException}, if yes it is returned in a Mono.error.
* Otherwise the content is returned in the Mono
*
* @param content the content to analyze
* @param mediaType the returned media type
* @param status the response status
* @return a Mono with the content or an Mono.error
*/
private static Mono<String> contentOrError(String content, String mediaType, RestStatus status) {
ElasticsearchException exception = getElasticsearchException(content, mediaType, status);
if (exception != null) {
StringBuilder sb = new StringBuilder();
buildExceptionMessages(sb, exception);
return Mono.error(new ElasticsearchStatusException(sb.toString(), status, exception));
}
return Mono.just(content);
}
/**
* tries to parse an {@link ElasticsearchException} from the given body content
*
* @param content the content to analyse
* @param mediaType the type of the body content
* @return an {@link ElasticsearchException} or {@literal null}.
*/
@Nullable @Nullable
private ElasticsearchException getElasticsearchException(ClientResponse response, String content, String mediaType) private static ElasticsearchException getElasticsearchException(String content, String mediaType, RestStatus status) {
throws IOException {
XContentParser parser = createParser(mediaType, content); try {
// we have a JSON object with an error and a status field XContentParser parser = createParser(mediaType, content);
XContentParser.Token token = parser.nextToken(); // Skip START_OBJECT // we have a JSON object with an error and a status field
XContentParser.Token token = parser.nextToken(); // Skip START_OBJECT
do { do {
token = parser.nextToken(); token = parser.nextToken();
if (parser.currentName().equals("error")) { if (parser.currentName().equals("error")) {
return ElasticsearchException.failureFromXContent(parser); return ElasticsearchException.failureFromXContent(parser);
} }
} while (token == XContentParser.Token.FIELD_NAME); } while (token == XContentParser.Token.FIELD_NAME);
return null;
return null;
} catch (IOException e) {
return new ElasticsearchStatusException(content, status);
}
} }
private static void buildExceptionMessages(StringBuilder sb, Throwable t) { private static void buildExceptionMessages(StringBuilder sb, Throwable t) {