src/java.net.http/share/classes/jdk/internal/net/http/Exchange.java - master - githubmirror/galahad

src/java.net.http/share/classes/jdk/internal/net/http/Exchange.java
776 строк34 KB
Jaikiran Pai
8326498: java.net.http.HttpClient connection leak using http/2
25 ноя 2025, 14:13
25 ноя 2025, 14:13c19b129
/*
 * Copyright (c) 2015, 2025, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

package jdk.internal.net.http;

import java.io.IOException;
import java.net.ProtocolException;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.net.http.HttpClient;
import java.net.http.HttpResponse;
import java.net.http.HttpTimeoutException;

import jdk.internal.net.http.HttpClientImpl.DelegatingExecutor;
import jdk.internal.net.http.common.Logger;
import jdk.internal.net.http.common.MinimalFuture;
import jdk.internal.net.http.common.Utils;
import jdk.internal.net.http.common.Log;

/**
 * One request/response exchange (handles 100/101 intermediate response also).
 * depth field used to track number of times a new request is being sent
 * for a given API request. If limit exceeded exception is thrown.
 *
 */
final class Exchange<T> {

    static final int MAX_NON_FINAL_RESPONSES =
            Utils.getIntegerNetProperty("jdk.httpclient.maxNonFinalResponses", 8);
    final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);

    final HttpRequestImpl request;
    final HttpClientImpl client;
    volatile ExchangeImpl<T> exchImpl;
    volatile CompletableFuture<? extends ExchangeImpl<T>> exchangeCF;
    volatile CompletableFuture<Void> bodyIgnored;
    volatile boolean streamLimitReached;

    // used to record possible cancellation raised before the exchImpl
    // has been established.
    private final AtomicReference<IOException> failed = new AtomicReference<>();
    final MultiExchange<T> multi;
    final DelegatingExecutor parentExecutor;
    volatile boolean upgrading; // to HTTP/2
    volatile boolean upgraded;  // to HTTP/2
    final PushGroup<T> pushGroup;
    final String dbgTag;

    // Keeps track of the underlying connection when establishing an HTTP/2
    // or HTTP/3 exchange so that it can be aborted/timed out mid-setup.
    final ConnectionAborter connectionAborter = new ConnectionAborter();

    final AtomicInteger nonFinalResponses = new AtomicInteger();

    // This will be set to true only when it is guaranteed that the server hasn't processed
    // the request. Typically, this happens when the server explicitly states (through a GOAWAY frame
    // or a relevant error code in reset frame) that the corresponding stream (id) wasn't processed.
    // However, there can be cases where the client is certain that the request wasn't sent
    // to the server (and thus not processed). In such cases, the client can set this to true.
    private volatile boolean unprocessedByPeer;

    Exchange(HttpRequestImpl request, MultiExchange<T> multi) {
        this.request = request;
        this.upgrading = false;
        this.client = multi.client();
        this.multi = multi;
        this.parentExecutor = multi.executor;
        this.pushGroup = multi.pushGroup;
        this.dbgTag = "Exchange";
    }

    PushGroup<T> getPushGroup() {
        return pushGroup;
    }

    DelegatingExecutor executor() {
        return parentExecutor;
    }

    public HttpRequestImpl request() {
        return request;
    }

    public Optional<Duration> remainingConnectTimeout() {
        return multi.remainingConnectTimeout();
    }

    HttpClientImpl client() {
        return client;
    }

    // Keeps track of the underlying connection when establishing an HTTP/2
    // or HTTP/3 exchange so that it can be aborted/timed out mid setup.
    final class ConnectionAborter {
        // In case of HTTP/3 requests we may have
        // two connections in parallel: a regular TCP connection
        // and a QUIC connection.
        private volatile HttpConnection connection;
        private volatile HttpQuicConnection quicConnection;
        private volatile boolean closeRequested;
        private volatile Throwable cause;

        void connection(HttpConnection connection) {
            boolean closeRequested;
            synchronized (this) {
                // check whether this new connection should be
                // closed
                closeRequested = this.closeRequested;
                if (!closeRequested) {
                    if (connection instanceof HttpQuicConnection quicConnection) {
                        this.quicConnection = quicConnection;
                    } else {
                        this.connection = connection;
                    }
                }
            }
            if (closeRequested) closeConnection(connection, cause);
        }

        void closeConnection(Throwable error) {
            HttpConnection connection;
            HttpQuicConnection quicConnection;
            Throwable cause;
            synchronized (this) {
                cause = this.cause;
                if (cause == null) {
                    cause = error;
                }
                connection = this.connection;
                quicConnection = this.quicConnection;
                if (connection == null || quicConnection == null) {
                    closeRequested = true;
                    this.cause = cause;
                } else {
                    this.quicConnection = null;
                    this.connection = null;
                    this.cause = null;
                }
            }
            closeConnection(connection, cause);
            closeConnection(quicConnection, cause);
        }

        // Called by HTTP/2 after an upgrade.
        // There is no upgrade for HTTP/3
        HttpConnection disable() {
            HttpConnection connection;
            synchronized (this) {
                connection = this.connection;
                this.connection = null;
                this.quicConnection = null;
                this.closeRequested = false;
                this.cause = null;
            }
            return connection;
        }

        void clear(HttpConnection connection) {
            synchronized (this) {
                var c = this.connection;
                if (connection == c) this.connection = null;
                var qc = this.quicConnection;
                if (connection == qc) this.quicConnection = null;
            }
        }

        private void closeConnection(HttpConnection connection, Throwable cause) {
            if (connection == null) {
                return;
            }
            try {
                connection.close(cause);
            } catch (Throwable t) {
                // ignore
                if (debug.on()) {
                    debug.log("ignoring exception that occurred during closing of connection: "
                            + connection, t);
                }
            }
        }
    }

    // true if previous attempt resulted in streamLimitReached
    public boolean hasReachedStreamLimit() { return streamLimitReached; }
    // can be used to set or clear streamLimitReached (for instance clear it after retrying)
    void streamLimitReached(boolean streamLimitReached) { this.streamLimitReached = streamLimitReached; }

    // Called for 204 response - when no body is permitted
    // This is actually only needed for HTTP/1.1 in order
    // to return the connection to the pool (or close it)
    void nullBody(HttpResponse<T> resp, Throwable t) {
        exchImpl.nullBody(resp, t);
    }

    public CompletableFuture<T> readBodyAsync(HttpResponse.BodyHandler<T> handler) {
        // If we received a 407 while establishing the exchange
        // there will be no body to read: bodyIgnored will be true,
        // and exchImpl will be null (if we were trying to establish
        // an HTTP/2 tunnel through an HTTP/1.1 proxy)
        if (bodyIgnored != null) return MinimalFuture.completedFuture(null);

        // The connection will not be returned to the pool in the case of WebSocket
        return exchImpl.readBodyAsync(handler, !request.isWebSocket(), parentExecutor)
                .whenComplete((r,t) -> exchImpl.completed());
    }

    /**
     * Called after a redirect or similar kind of retry where a body might
     * be sent but we don't want it. Should send a RESET in h2. For http/1.1
     * we can consume small quantity of data, or close the connection in
     * other cases.
     */
    public CompletableFuture<Void> ignoreBody() {
        if (bodyIgnored != null) return bodyIgnored;
        return exchImpl.ignoreBody();
    }

    /**
     * Called when a new exchange is created to replace this exchange.
     * At this point it is guaranteed that readBody/readBodyAsync will
     * not be called.
     */
    public void released() {
        ExchangeImpl<?> impl = exchImpl;
        if (impl != null) impl.released();
        // Don't set exchImpl to null here. We need to keep
        // it alive until it's replaced by a Stream in wrapForUpgrade.
        // Setting it to null here might get it GC'ed too early, because
        // the Http1Response is now only weakly referenced by the Selector.
    }

    public void cancel() {
        // cancel can be called concurrently before or at the same time
        // that the exchange impl is being established.
        // In that case we won't be able to propagate the cancellation
        // right away
        if (exchImpl != null) {
            exchImpl.cancel();
        } else {
            // no impl - can't cancel impl yet.
            // call cancel(IOException) instead which takes care
            // of race conditions between impl/cancel.
            cancel(new IOException("Request cancelled"));
        }
    }

    public void cancel(IOException cause) {
        if (debug.on()) debug.log("cancel exchImpl: %s, with \"%s\"", exchImpl, cause);
        // If the impl is non null, propagate the exception right away.
        // Otherwise record it so that it can be propagated once the
        // exchange impl has been established.
        ExchangeImpl<?> impl;
        IOException closeReason = null;
        synchronized (this) {
            impl =  exchImpl;
            if (impl == null) {
                // no impl yet. record the exception
                failed.compareAndSet(null, cause);
            }
        }
        if (impl != null) {
            // propagate the exception to the impl
            if (debug.on()) debug.log("Cancelling exchImpl: %s", exchImpl);
            impl.cancel(cause);
        } else {
             // abort/close the connection if setting up the exchange. This can
            // be important when setting up HTTP/2 or HTTP/3
            closeReason = failed.get();
            if (closeReason != null) {
                connectionAborter.closeConnection(closeReason);
            }

            // now call checkCancelled to recheck the impl.
            // if the failed state is set and the impl is not null, reset
            // the failed state and propagate the exception to the impl.
            checkCancelled();
        }
    }

    // This method will raise an exception if one was reported and if
    // it is possible to do so. If the exception can be raised, then
    // the failed state will be reset. Otherwise, the failed state
    // will persist until the exception can be raised and the failed state
    // can be cleared.
    // Takes care of possible race conditions.
    private void checkCancelled() {
        ExchangeImpl<?> impl = null;
        IOException cause = null;
        CompletableFuture<? extends ExchangeImpl<T>> cf = null;
        if (failed.get() != null) {
            synchronized (this) {
                cause = failed.get();
                impl = exchImpl;
                cf = exchangeCF;
            }
        }
        if (multi.requestCancelled() && impl != null && cause == null) {
            cause = new IOException("Request cancelled");
        }
        if (cause == null) return;
        if (impl != null) {
            // The exception is raised by propagating it to the impl.
            if (debug.on()) debug.log("Cancelling exchImpl: %s", impl);
            impl.cancel(cause);
            synchronized (this) {
                if (impl == exchImpl) {
                    failed.compareAndSet(cause, null);
                }
            }
        } else {
            Log.logTrace("Exchange: request [{0}/timeout={1}ms] no impl is set."
                         + "\n\tCan''t cancel yet with {2}",
                         request.uri(),
                         request.timeout().isPresent() ?
                         // calling duration.toMillis() can throw an exception.
                         // this is just debugging, we don't care if it overflows.
                         (request.timeout().get().getSeconds() * 1000
                          + request.timeout().get().getNano() / 1000000) : -1,
                         cause);
            if (cf != null) cf.completeExceptionally(cause);
        }
    }

    <U> CompletableFuture<U> checkCancelled(CompletableFuture<U> cf, HttpConnection connection) {
        return cf.handle((r,t) -> {
            if (t == null) {
                if (multi.requestCancelled()) {
                    // if upgraded, we don't close the connection.
                    // cancelling will be handled by the HTTP/2 exchange
                    // in its own time.
                    if (!upgraded && !(connection instanceof HttpQuicConnection)) {
                        t = getCancelCause();
                        if (t == null) t = new IOException("Request cancelled");
                        if (debug.on()) debug.log("exchange cancelled during connect: " + t);
                        try {
                            connection.close(t);
                        } catch (Throwable x) {
                            if (debug.on()) debug.log("Failed to close connection", x);
                        }
                        return MinimalFuture.<U>failedFuture(t);
                    }
                }
            }
            return cf;
        }).thenCompose(Function.identity());
    }

    public void h2Upgrade() {
        upgrading = true;
        request.setH2Upgrade(this);
    }

    synchronized IOException failed(IOException io) {
        IOException cause = failed.compareAndExchange(null, io);
        return cause == null ? io : cause;
    }

    synchronized IOException getCancelCause() {
        return failed.get();
    }

    // get/set the exchange impl, solving race condition issues with
    // potential concurrent calls to cancel() or cancel(IOException)
    private CompletableFuture<? extends ExchangeImpl<T>>
    establishExchange(HttpConnection connection) {
        if (debug.on()) {
            debug.log("establishing exchange for %s #%s,%n\t proxy=%s",
                      request, multi.id, request.proxy());
        }
        // check if we have been cancelled first.
        Throwable t = getCancelCause();
        checkCancelled();
        if (t != null) {
            if (debug.on()) {
                debug.log("exchange was cancelled: returned failed cf (%s)", String.valueOf(t));
            }
            return exchangeCF = MinimalFuture.failedFuture(t);
        }

        CompletableFuture<? extends ExchangeImpl<T>> cf, res;

        cf = ExchangeImpl.get(this, connection)
                // set exchImpl and call checkCancelled to make  sure exchImpl
                // gets cancelled even if the exchangeCf was completed exceptionally
                // before the CF returned by ExchangeImpl.get completed. This deals
                // with issues when the request is cancelled while the exchange impl
                // is being created.
                .thenApply((eimpl) -> {
                    synchronized (Exchange.this) {exchImpl = eimpl;}
                    checkCancelled(); return eimpl;
                }).copy();
        // We should probably use a VarHandle to get/set exchangeCF
        // instead - as we need CAS semantics.
        synchronized (this) { exchangeCF = cf; };
        res = cf.whenComplete((r,x) -> {
            synchronized (Exchange.this) {
                if (exchangeCF == cf) exchangeCF = null;
            }
        });
        checkCancelled();
        return res.thenCompose((eimpl) -> {
                    // recheck for cancelled, in case of race conditions
                    exchImpl = eimpl;
                    IOException tt = getCancelCause();
                    checkCancelled();
                    if (tt != null) {
                        return MinimalFuture.failedFuture(tt);
                    } else {
                        // Now we're good to go. Because exchImpl is no longer
                        // null cancel() will be able to propagate directly to
                        // the impl after this point ( if needed ).
                        return MinimalFuture.completedFuture(eimpl);
                    } });
    }

    // Completed HttpResponse will be null if response succeeded
    // will be a non-null responseAsync if expect continue returns an error

    public CompletableFuture<Response> responseAsync() {
        return responseAsyncImpl(null);
    }

    // check whether the headersSentCF was completed exceptionally with
    // ProxyAuthorizationRequired. If so the Response embedded in the
    // exception is returned. Otherwise we proceed.
    private CompletableFuture<Response> checkFor407(ExchangeImpl<T> ex, Throwable t,
                                                    Function<ExchangeImpl<T>,CompletableFuture<Response>> andThen) {
        t = Utils.getCompletionCause(t);
        if (t instanceof ProxyAuthenticationRequired) {
            if (debug.on()) debug.log("checkFor407: ProxyAuthenticationRequired: building synthetic response");
            bodyIgnored = MinimalFuture.completedFuture(null);
            Response proxyResponse = ((ProxyAuthenticationRequired)t).proxyResponse;
            HttpConnection c = ex == null ? null : ex.connection();
            Response syntheticResponse = new Response(request, this,
                    proxyResponse.headers, c, proxyResponse.statusCode,
                    proxyResponse.version, true);
            return MinimalFuture.completedFuture(syntheticResponse);
        } else if (t != null) {
            if (debug.on()) debug.log("checkFor407: no response - %s", (Object)t);
            return MinimalFuture.failedFuture(t);
        } else {
            if (debug.on()) debug.log("checkFor407: all clear");
            return andThen.apply(ex);
        }
    }

    private CompletableFuture<Response> startSendingBody(DelegatingExecutor executor) {
        return exchImpl.sendBodyAsync()
                        .thenCompose(exIm -> exIm.getResponseAsync(executor));
    }

    // After sending the request headers, if no ProxyAuthorizationRequired
    // was raised and the expectContinue flag is on, we need to wait
    // for the 100-Continue response
    private CompletableFuture<Response> expectContinue(ExchangeImpl<T> ex) {
        assert request.expectContinue();

        long responseTimeoutMillis = 5000;
        if (request.timeout().isPresent()) {
            final long timeoutMillis = request.timeout().get().toMillis();
            responseTimeoutMillis = Math.min(responseTimeoutMillis, timeoutMillis);
        }

        return ex.getResponseAsync(parentExecutor)
                .completeOnTimeout(null, responseTimeoutMillis, TimeUnit.MILLISECONDS)
                .thenCompose((Response r1) -> {
                    // The response will only be null if there was a timeout
                    // send body regardless
                    if (r1 == null) {
                        if (debug.on())
                            debug.log("Setting ExpectTimeoutRaised and sending request body");
                        exchImpl.setExpectTimeoutRaised();
                        CompletableFuture<Response> cf = startSendingBody(parentExecutor);
                        cf = wrapForUpgrade(cf);
                        cf = wrapForLog(cf);
                        return cf;
                    }

                    Log.logResponse(r1::toString);
                    int rcode = r1.statusCode();
                    if (rcode == 100) {
                        nonFinalResponses.incrementAndGet();
                        Log.logTrace("Received 100-Continue: sending body");
                        if (debug.on()) debug.log("Received 100-Continue for %s", r1);
                        CompletableFuture<Response> cf = startSendingBody(parentExecutor);
                        cf = wrapForUpgrade(cf);
                        cf = wrapForLog(cf);
                        return cf;
                    } else {
                        Log.logTrace("Expectation failed: Received {0}",
                                rcode);
                        if (debug.on()) debug.log("Expect-Continue failed (%d) for: %s", rcode, r1);
                        if (upgrading && rcode == 101) {
                            IOException failed = new IOException(
                                    "Unable to handle 101 while waiting for 100");
                            return MinimalFuture.failedFuture(failed);
                        }
                        exchImpl.expectContinueFailed(rcode);
                        return MinimalFuture.completedFuture(r1);
                    }
                });
    }

    // After sending the request headers, if no ProxyAuthorizationRequired
    // was raised and the expectContinue flag is off, we can immediately
    // send the request body and proceed.
    private CompletableFuture<Response> sendRequestBody(ExchangeImpl<T> ex) {
        assert !request.expectContinue();
        if (debug.on()) debug.log("sendRequestBody");
        CompletableFuture<Response> cf = startSendingBody(parentExecutor);
        cf = wrapForUpgrade(cf);
        // after 101 is handled we check for other 1xx responses
        cf = cf.thenCompose(this::ignore1xxResponse);
        cf = wrapForLog(cf);
        return cf;
    }

    /**
     * Checks whether the passed Response has a status code between 102 and 199 (both inclusive).
     * If so, then that {@code Response} is considered intermediate informational response and is
     * ignored by the client. This method then creates a new {@link CompletableFuture} which
     * completes when a subsequent response is sent by the server. Such newly constructed
     * {@link CompletableFuture} will not complete till a "final" response (one which doesn't have
     * a response code between 102 and 199 inclusive) is sent by the server. The returned
     * {@link CompletableFuture} is thus capable of handling multiple subsequent intermediate
     * informational responses from the server.
     * <p>
     * If the passed Response doesn't have a status code between 102 and 199 (both inclusive) then
     * this method immediately returns back a completed {@link CompletableFuture} with the passed
     * {@code Response}.
     * </p>
     *
     * @param rsp The response
     * @return A {@code CompletableFuture} with the final response from the server
     */
    private CompletableFuture<Response> ignore1xxResponse(final Response rsp) {
        final int statusCode = rsp.statusCode();
        // we ignore any response code which is 1xx.
        // For 100 (with the request configured to expect-continue) and 101, we handle it
        // specifically as defined in the RFC-9110, outside of this method.
        // As noted in RFC-9110, section 15.2.1, if response code is 100 and if the request wasn't
        // configured with expectContinue, then we ignore the 100 response and wait for the final
        // response (just like any other 1xx response).
        // Any other response code between 102 and 199 (both inclusive) aren't specified in the
        // "HTTP semantics" RFC-9110. The spec states that these 1xx response codes are informational
        // and interim and the client can choose to ignore them and continue to wait for the
        // final response (headers)
        if ((statusCode >= 102 && statusCode <= 199)
                || (statusCode == 100 && !request.expectContinue)) {
            Log.logTrace("Ignoring (1xx informational) response code {0}", rsp.statusCode());
            if (debug.on()) {
                debug.log("Ignoring (1xx informational) response code "
                        + rsp.statusCode());
            }
            assert exchImpl != null : "Illegal state - current exchange isn't set";
            int count = nonFinalResponses.incrementAndGet();
            if (MAX_NON_FINAL_RESPONSES > 0 && (count < 0 || count > MAX_NON_FINAL_RESPONSES)) {
                return MinimalFuture.failedFuture(
                        new ProtocolException(String.format(
                                "Too many interim responses received: %s > %s",
                                count, MAX_NON_FINAL_RESPONSES)));
            } else {
                // ignore this Response and wait again for the subsequent response headers
                final CompletableFuture<Response> cf = exchImpl.getResponseAsync(parentExecutor);
                // we recompose the CF again into the ignore1xxResponse check/function because
                // the 1xx response is allowed to be sent multiple times for a request, before
                // a final response arrives
                return cf.thenCompose(this::ignore1xxResponse);
            }
        } else {
            // return the already completed future
            return MinimalFuture.completedFuture(rsp);
        }
    }

    CompletableFuture<Response> responseAsyncImpl(HttpConnection connection) {
        Function<ExchangeImpl<T>, CompletableFuture<Response>> after407Check;
        bodyIgnored = null;
        if (request.expectContinue()) {
            request.setSystemHeader("Expect", "100-Continue");
            Log.logTrace("Sending Expect: 100-Continue");
            // wait for 100-Continue before sending body
            after407Check = this::expectContinue;
        } else {
            // send request body and proceed.
            after407Check = this::sendRequestBody;
        }
        // The ProxyAuthorizationRequired can be triggered either by
        // establishExchange (case of HTTP/2 SSL tunneling through HTTP/1.1 proxy
        // or by sendHeaderAsync (case of HTTP/1.1 SSL tunneling through HTTP/1.1 proxy
        // Therefore we handle it with a call to this checkFor407(...) after these
        // two places.
        Function<ExchangeImpl<T>, CompletableFuture<Response>> afterExch407Check =
                (ex) -> ex.sendHeadersAsync()
                        .handle((r,t) -> this.checkFor407(r, t, after407Check))
                        .thenCompose(Function.identity());
        return establishExchange(connection)
                .handle((r,t) -> this.checkFor407(r,t, afterExch407Check))
                .thenCompose(Function.identity());
    }

    private CompletableFuture<Response> wrapForUpgrade(CompletableFuture<Response> cf) {
        if (upgrading) {
            return cf.thenCompose(r -> checkForUpgradeAsync(r, exchImpl));
        }
        // websocket requests use "Connection: Upgrade" and "Upgrade: websocket" headers.
        // however, the "upgrading" flag we maintain in this class only tracks a h2 upgrade
        // that we internally triggered. So it will be false in the case of websocket upgrade, hence
        // this additional check. If it's a websocket request we allow 101 responses and we don't
        // require any additional checks when a response arrives.
        if (request.isWebSocket()) {
            return cf;
        }
        // not expecting an upgrade, but if the server sends a 101 response then we fail the
        // request and also let the ExchangeImpl deal with it as a protocol error
        return cf.thenCompose(r -> {
            if (r.statusCode == 101) {
                final ProtocolException protoEx = new ProtocolException("Unexpected 101 " +
                        "response, when not upgrading");
                assert exchImpl != null : "Illegal state - current exchange isn't set";
                try {
                    exchImpl.onProtocolError(protoEx);
                } catch (Throwable ignore){
                    // ignored
                }
                return MinimalFuture.failedFuture(protoEx);
            }
            return MinimalFuture.completedFuture(r);
        });
    }

    private CompletableFuture<Response> wrapForLog(CompletableFuture<Response> cf) {
        if (Log.requests()) {
            return cf.thenApply(response -> {
                Log.logResponse(response::toString);
                return response;
            });
        }
        return cf;
    }

    HttpResponse.BodySubscriber<T> ignoreBody(HttpResponse.ResponseInfo hdrs) {
        return HttpResponse.BodySubscribers.replacing(null);
    }

    // if this response was received in reply to an upgrade
    // then create the Http2Connection from the HttpConnection
    // initialize it and wait for the real response on a newly created Stream

    private CompletableFuture<Response>
    checkForUpgradeAsync(Response resp,
                         ExchangeImpl<T> ex) {

        int rcode = resp.statusCode();
        if (upgrading && (rcode == 101)) {
            Http1Exchange<T> e = (Http1Exchange<T>)ex;
            // check for 101 switching protocols
            // 101 responses are not supposed to contain a body.
            //    => should we fail if there is one?
            if (debug.on()) debug.log("Upgrading async %s", e.connection());
            return e.readBodyAsync(this::ignoreBody, false, parentExecutor)
                .thenCompose((T v) -> {// v is null
                    debug.log("Ignored body");
                    // we pass e::getBuffer to allow the ByteBuffers to accumulate
                    // while we build the Http2Connection
                    ex.upgraded();
                    upgraded = true;
                    return Http2Connection.createAsync(e.connection(),
                                                 client.client2(),
                                                 this, e::drainLeftOverBytes)
                        .thenCompose((Http2Connection c) -> {
                            HttpConnection connection = connectionAborter.disable();
                            boolean cached = c.offerConnection();
                            if (!cached && connection != null) {
                                connectionAborter.connection(connection);
                            }
                            Stream<T> s = c.getInitialStream();

                            if (s == null) {
                                // s can be null if an exception occurred
                                // asynchronously while sending the preface.
                                final Http2TerminationCause tc = c.getTerminationCause();
                                IOException ioe;
                                if (tc != null) {
                                    ioe = new IOException("Can't get stream 1", tc.getCloseCause());
                                } else {
                                    ioe = new IOException("Can't get stream 1");
                                }
                                return MinimalFuture.failedFuture(ioe);
                            }
                            exchImpl.released();
                            Throwable t;
                            // There's a race condition window where an external
                            // thread (SelectorManager) might complete the
                            // exchange in timeout at the same time where we're
                            // trying to switch the exchange impl.
                            // 'failed' will be reset to null after
                            // exchImpl.cancel() has completed, so either we
                            // will observe failed != null here, or we will
                            // observe e.getCancelCause() != null, or the
                            // timeout exception will be routed to 's'.
                            // Either way, we need to relay it to s.
                            synchronized (this) {
                                exchImpl = s;
                                t = failed.get();
                            }
                            // Check whether the HTTP/1.1 was cancelled.
                            if (t == null) t = e.getCancelCause();
                            // if HTTP/1.1 exchange was timed out, or the request
                            // was cancelled don't try to go further.
                            if (t instanceof HttpTimeoutException || multi.requestCancelled()) {
                                if (t == null) t = new IOException("Request cancelled");
                                s.cancelImpl(t);
                                return MinimalFuture.failedFuture(t);
                            }
                            if (debug.on())
                                debug.log("Getting response async %s", s);
                            return s.getResponseAsync(null);
                        });}
                );
        }
        return MinimalFuture.completedFuture(resp);
    }

    HttpClient.Version version() {
        return multi.version();
    }

    boolean pushEnabled() {
        return pushGroup != null;
    }

    String h2cSettingsStrings() {
        return client.client2().getSettingsString(pushEnabled());
    }

    String dbgString() {
        return dbgTag;
    }

    final boolean isUnprocessedByPeer() {
        return this.unprocessedByPeer;
    }

    // Marks the exchange as unprocessed by the peer
    final void markUnprocessedByPeer() {
        this.unprocessedByPeer = true;
    }
}