| /* |
| * 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.Executor; |
| 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; |
| |
| // 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 |
| // exchange so that it can be aborted/timed out mid setup. |
| final ConnectionAborter connectionAborter = new ConnectionAborter(); |
| |
| final AtomicInteger nonFinalResponses = new AtomicInteger(); |
| |
| 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 |
| // exchange so that it can be aborted/timed out mid setup. |
| static final class ConnectionAborter { |
| private volatile HttpConnection connection; |
| 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) { |
| this.connection = connection; |
| } else { |
| // assert this.connection == null |
| this.closeRequested = false; |
| } |
| } |
| if (closeRequested) closeConnection(connection, cause); |
| } |
| |
| void closeConnection(Throwable error) { |
| HttpConnection connection; |
| Throwable cause; |
| synchronized (this) { |
| cause = this.cause; |
| if (cause == null) { |
| cause = error; |
| } |
| connection = this.connection; |
| if (connection == null) { |
| closeRequested = true; |
| this.cause = cause; |
| } else { |
| this.connection = null; |
| this.cause = null; |
| } |
| } |
| closeConnection(connection, cause); |
| } |
| |
| HttpConnection disable() { |
| HttpConnection connection; |
| synchronized (this) { |
| connection = this.connection; |
| this.connection = null; |
| this.closeRequested = false; |
| this.cause = null; |
| } |
| return connection; |
| } |
| |
| private static void closeConnection(HttpConnection connection, Throwable cause) { |
| if (connection != null) { |
| try { |
| connection.close(cause); |
| } catch (Throwable t) { |
| // ignore |
| } |
| } |
| } |
| } |
| |
| // 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 |
| 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 (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) { |
| 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,%n\t proxy=%s", |
| request, 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); |
| // 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. |
| Throwable t = c.getRecordedCause(); |
| IOException ioe; |
| if (t != null) { |
| if (!cached) |
| c.close(); |
| ioe = new IOException("Can't get stream 1: " + t, t); |
| } 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; |
| } |
| } |
