Как прервать Flux после 3 ошибок ⇐ JAVA

[Anonymous]

Цель:

• получить заголовки ext org.springframework.http.HttpHeaders с помощью org.springframework.web.reactive.function.client.WebClient из сгенерированного списка конечных точек, не зная их точного количества.
• ограничить постоянное количество запросов до появления 3 ошибок
  например. https://source-host/resolve/enpoint/R000 на R200 и удалите все Mono.empty()

Код: Выделить всё

    public Mono askForRedirectHeaders(String urlBase, String token, List paths) {
WebClient clientBase = getClientBase(urlBase, token);
Flux headersListByEndPathsExt = clientBase.getHeadersListByEndPathsLimited(paths);
return headersListByEndPathsExt != null ? headersListByEndPathsExt.onErrorResume(error -> Mono.empty())
.collectMap(HttpHeadersExt::getPathExt, this::recognizeRedirection)
: Mono.empty();

public Flux getHeadersListByEndPathsLimited(List endPaths) {
return Flux.fromIterable(endPaths).filter(n -> !n.isEmpty())
.flatMap(this::getHeadersByEndPathLimited)
.doOnError(it -> errorLog(it, log));
}

public Mono getHeadersByEndPathLimited(String endPath) {
return this.webClient.head().uri("/{endPath}", endPath).exchangeToMono(response -> {
if (response.statusCode().isError()) {
String description = getResponseDescription(response);
return Mono.error(new PathUnavailableException(description));
} else {
return response.toEntity(String.class).map(it -> new HttpHeadersExt(it.getHeaders(), endPath));
}
});
}

но doOnError немедленно прерывает все незавершенные соединения, и результат - ex. «R002», «R006», «R009», «R015».
Поток должен быть построен с использованием результатов из упр. Mono.empty() для «R000», «R001», «R002», «R003», «R004», «R005» и т. д. до еще двух Mono.empty().

Подробнее здесь: https://stackoverflow.com/questions/798 ... r-3-errors