HttpClient extends ClientTransport<HttpClient, HttpClientConfig> { public final HttpClient doOnError(BiConsumer<? super HttpClientRequest, ? super Throwable> doOnRequestError, BiConsumer<? super HttpClientResponse, ? super Throwable> doOnResponseError) { Objects.requireNonNull(doOnRequestError, "doOnRequestError"); Objects.requireNonNull(doOnResponseError, "doOnResponseError"); HttpClient dup = duplicate(); @SuppressWarnings("unchecked") BiConsumer<HttpClientRequest, Throwable> currentRequestError = (BiConsumer<HttpClientRequest, Throwable>) configuration().doOnRequestError; dup.configuration().doOnRequestError = currentRequestError == null ? doOnRequestError : currentRequestError.andThen(doOnRequestError); @SuppressWarnings("unchecked") BiConsumer<HttpClientResponse, Throwable> currentResponseError = (BiConsumer<HttpClientResponse, Throwable>) configuration().doOnResponseError; dup.configuration().doOnResponseError = currentResponseError == null ? doOnResponseError : currentResponseError.andThen(doOnResponseError); return dup; }  static HttpClient create(); static HttpClient create(ConnectionProvider connectionProvider); static HttpClient create(ConnectionProvider connectionProvider, int maxHttp2Connections); @Deprecated static HttpClient from(TcpClient tcpClient); static HttpClient newConnection(); final HttpClient baseUrl(String baseUrl); final HttpClient compress(boolean compressionEnabled); final HttpClient cookie(Cookie cookie); final HttpClient cookie(String name, Consumer<? super Cookie> cookieBuilder); final HttpClient cookieCodec(ClientCookieEncoder encoder); final HttpClient cookieCodec(ClientCookieEncoder encoder, ClientCookieDecoder decoder); final HttpClient cookiesWhen(String name, Function<? super Cookie, Mono<? extends Cookie>> cookieBuilder); final RequestSender delete(); final HttpClient disableRetry(boolean disableRetry); final HttpClient doAfterRequest(BiConsumer<? super HttpClientRequest, ? super Connection> doAfterRequest); final HttpClient doAfterResponseSuccess(BiConsumer<? super HttpClientResponse, ? super Connection> doAfterResponseSuccess); final HttpClient doOnError(BiConsumer<? super HttpClientRequest, ? super Throwable> doOnRequestError,
			BiConsumer<? super HttpClientResponse, ? super Throwable> doOnResponseError); final HttpClient doOnRedirect(BiConsumer<? super HttpClientResponse, ? super Connection> doOnRedirect); final HttpClient doOnRequest(BiConsumer<? super HttpClientRequest, ? super Connection> doOnRequest); final HttpClient doOnRequestError(BiConsumer<? super HttpClientRequest, ? super Throwable> doOnRequestError); final HttpClient doOnResponse(BiConsumer<? super HttpClientResponse, ? super Connection> doOnResponse); final HttpClient doOnResponseError(BiConsumer<? super HttpClientResponse, ? super Throwable> doOnResponseError); final HttpClient followRedirect(BiPredicate<HttpClientRequest, HttpClientResponse> predicate); final HttpClient followRedirect(BiPredicate<HttpClientRequest, HttpClientResponse> predicate,
			@Nullable BiConsumer<HttpHeaders, HttpClientRequest> redirectRequestBiConsumer); final HttpClient followRedirect(BiPredicate<HttpClientRequest, HttpClientResponse> predicate,
			@Nullable Consumer<HttpClientRequest> redirectRequestConsumer); final HttpClient followRedirect(boolean followRedirect); final HttpClient followRedirect(boolean followRedirect,
			@Nullable BiConsumer<HttpHeaders, HttpClientRequest> redirectRequestBiConsumer); final HttpClient followRedirect(boolean followRedirect, @Nullable Consumer<HttpClientRequest> redirectRequestConsumer); final ResponseReceiver<?> get(); final ResponseReceiver<?> head(); final HttpClient headers(Consumer<? super HttpHeaders> headerBuilder); final HttpClient headersWhen(Function<? super HttpHeaders, Mono<? extends HttpHeaders>> headerBuilder); final HttpClient http2Settings(Consumer<Http2SettingsSpec.Builder> http2Settings); final HttpClient httpResponseDecoder(Function<HttpResponseDecoderSpec, HttpResponseDecoderSpec> responseDecoderOptions); final HttpClient keepAlive(boolean keepAlive); final HttpClient mapConnect(
			Function<? super Mono<? extends Connection>, ? extends Mono<? extends Connection>> connector); final HttpClient metrics(boolean enable, Function<String, String> uriTagValue); @Override final HttpClient metrics(boolean enable, Supplier<? extends ChannelMetricsRecorder> recorder); final HttpClient metrics(boolean enable, Supplier<? extends ChannelMetricsRecorder> recorder, Function<String, String> uriValue); final HttpClient noSSL(); @Override final HttpClient observe(ConnectionObserver observer); final ResponseReceiver<?> options(); final RequestSender patch(); @Override final HttpClient port(int port); final RequestSender post(); final HttpClient protocol(HttpProtocol... supportedProtocols); final RequestSender put(); @Override final HttpClient remoteAddress(Supplier<? extends SocketAddress> remoteAddressSupplier); RequestSender request(HttpMethod method); final HttpClient responseTimeout(Duration timeout); final HttpClient secure(); final HttpClient secure(Consumer<? super SslProvider.SslContextSpec> sslProviderBuilder); HttpClient secure(SslProvider sslProvider); @Deprecated @SuppressWarnings("ReturnValueIgnored") final HttpClient tcpConfiguration(Function<? super TcpClient, ? extends TcpClient> tcpMapper); final WebsocketSender websocket(); final WebsocketSender websocket(WebsocketClientSpec websocketClientSpec); @Override final HttpClient wiretap(boolean enable); static final String USER_AGENT; }
@Test public void doOnError() { disposableServer = HttpServer.create() .port(0) .handle((req, resp) -> { if (req.requestHeaders().contains("during")) { return resp.sendString(Flux.just("test").hide()) .then(Mono.error(new RuntimeException("test"))); } throw new RuntimeException("test"); }) .bindNow(); AtomicReference<String> requestError1 = new AtomicReference<>(); AtomicReference<String> responseError1 = new AtomicReference<>(); Mono<String> content = createHttpClientForContextWithPort() .headers(h -> h.add("before", "test")) .doOnRequestError((req, err) -> requestError1.set(req.currentContextView().getOrDefault("test", "empty"))) .doOnResponseError((res, err) -> responseError1.set(res.currentContextView().getOrDefault("test", "empty"))) .mapConnect((c) -> c.contextWrite(Context.of("test", "success"))) .get() .uri("/") .responseContent() .aggregate() .asString(); StepVerifier.create(content) .verifyError(PrematureCloseException.class); assertThat(requestError1.get()).isEqualTo("success"); assertThat(responseError1.get()).isNull(); AtomicReference<String> requestError2 = new AtomicReference<>(); AtomicReference<String> responseError2 = new AtomicReference<>(); content = createHttpClientForContextWithPort() .headers(h -> h.add("during", "test")) .doOnError((req, err) -> requestError2.set(req.currentContextView().getOrDefault("test", "empty")) ,(res, err) -> responseError2.set(res.currentContextView().getOrDefault("test", "empty"))) .mapConnect((c) -> c.contextWrite(Context.of("test", "success"))) .get() .uri("/") .responseContent() .aggregate() .asString(); StepVerifier.create(content) .verifyError(PrematureCloseException.class); assertThat(requestError2.get()).isNull(); assertThat(responseError2.get()).isEqualTo("success"); }