package jdk.internal.net.http;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.ConnectException;
import java.net.http.HttpConnectTimeoutException;
import java.util.Iterator;
import java.util.LinkedList;
import java.security.AccessControlContext;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Flow;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.net.http.HttpClient;
import java.net.http.HttpHeaders;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodySubscriber;
import java.net.http.HttpResponse.PushPromiseHandler;
import java.net.http.HttpTimeoutException;
import jdk.internal.net.http.common.Log;
import jdk.internal.net.http.common.Logger;
import jdk.internal.net.http.common.MinimalFuture;
import jdk.internal.net.http.common.ConnectionExpiredException;
import jdk.internal.net.http.common.Utils;
import static jdk.internal.net.http.common.MinimalFuture.completedFuture;
import static jdk.internal.net.http.common.MinimalFuture.failedFuture;
class MultiExchange<T> {
static final Logger debug =
Utils.getDebugLogger("MultiExchange"::toString, Utils.DEBUG);
private final HttpRequest userRequest;
private final HttpRequestImpl request;
final AccessControlContext acc;
final HttpClientImpl client;
final HttpResponse.BodyHandler<T> responseHandler;
final HttpClientImpl.DelegatingExecutor executor;
final AtomicInteger attempts = new AtomicInteger();
HttpRequestImpl currentreq;
HttpRequestImpl previousreq;
Exchange<T> exchange;
Exchange<T> previous;
volatile Throwable retryCause;
volatile boolean expiredOnce;
volatile HttpResponse<T> response = null;
static final int DEFAULT_MAX_ATTEMPTS = 5;
static final int max_attempts = Utils.getIntegerNetProperty(
"jdk.httpclient.redirects.retrylimit", DEFAULT_MAX_ATTEMPTS
);
private final LinkedList<HeaderFilter> filters;
ResponseTimerEvent responseTimerEvent;
volatile boolean cancelled;
final PushGroup<T> pushGroup;
volatile AuthenticationFilter.AuthInfo serverauth, proxyauth;
volatile int numberOfRedirects = 0;
MultiExchange(HttpRequest userRequest,
HttpRequestImpl requestImpl,
HttpClientImpl client,
HttpResponse.BodyHandler<T> responseHandler,
PushPromiseHandler<T> pushPromiseHandler,
AccessControlContext acc) {
this.previous = null;
this.userRequest = userRequest;
this.request = requestImpl;
this.currentreq = request;
this.previousreq = null;
this.client = client;
this.filters = client.filterChain();
this.acc = acc;
this.executor = client.theExecutor();
this.responseHandler = responseHandler;
if (pushPromiseHandler != null) {
Executor executor = acc == null
? this.executor.delegate()
: new PrivilegedExecutor(this.executor.delegate(), acc);
this.pushGroup = new PushGroup<>(pushPromiseHandler, request, executor);
} else {
pushGroup = null;
}
this.exchange = new Exchange<>(request, this);
}
synchronized Exchange<T> getExchange() {
return exchange;
}
HttpClientImpl client() {
return client;
}
HttpClient.Version version() {
HttpClient.Version vers = request.version().orElse(client.version());
if (vers == HttpClient.Version.HTTP_2 && !request.secure() && request.proxy() != null)
vers = HttpClient.Version.HTTP_1_1;
return vers;
}
private synchronized void setExchange(Exchange<T> exchange) {
if (this.exchange != null && exchange != this.exchange) {
this.exchange.released();
}
this.exchange = exchange;
}
private void cancelTimer() {
if (responseTimerEvent != null) {
client.cancelTimer(responseTimerEvent);
}
}
private void requestFilters(HttpRequestImpl r) throws IOException {
Log.logTrace("Applying request filters");
for (HeaderFilter filter : filters) {
Log.logTrace("Applying {0}", filter);
filter.request(r, this);
}
Log.logTrace("All filters applied");
}
private HttpRequestImpl responseFilters(Response response) throws IOException
{
Log.logTrace("Applying response filters");
Iterator<HeaderFilter> reverseItr = filters.descendingIterator();
while (reverseItr.hasNext()) {
HeaderFilter filter = reverseItr.next();
Log.logTrace("Applying {0}", filter);
HttpRequestImpl newreq = filter.response(response);
if (newreq != null) {
Log.logTrace("New request: stopping filters");
return newreq;
}
}
Log.logTrace("All filters applied");
return null;
}
public void cancel(IOException cause) {
cancelled = true;
getExchange().cancel(cause);
}
public CompletableFuture<HttpResponse<T>> responseAsync(Executor executor) {
CompletableFuture<Void> start = new MinimalFuture<>();
CompletableFuture<HttpResponse<T>> cf = responseAsync0(start);
start.completeAsync( () -> null, executor);
return cf;
}
private static boolean bodyNotPermitted(Response r) {
return r.statusCode == 204;
}
private boolean bodyIsPresent(Response r) {
HttpHeaders headers = r.headers();
if (headers.firstValueAsLong("Content-length").orElse(0L) != 0L)
return true;
if (headers.firstValue("Transfer-encoding").isPresent())
return true;
return false;
}
private CompletableFuture<HttpResponse<T>> handleNoBody(Response r, Exchange<T> exch) {
BodySubscriber<T> bs = responseHandler.apply(new ResponseInfoImpl(r.statusCode(),
r.headers(), r.version()));
CompletionStage<T> cs = bs.getBody();
bs.onSubscribe(new NullSubscription());
bs.onComplete();
MinimalFuture<HttpResponse<T>> result = new MinimalFuture<>();
cs.whenComplete((nullBody, exception) -> {
if (exception != null)
result.completeExceptionally(exception);
else {
this.response =
new HttpResponseImpl<>(r.request(), r, this.response, nullBody, exch);
result.complete(this.response);
}
});
return result;
}
private CompletableFuture<HttpResponse<T>>
responseAsync0(CompletableFuture<Void> start) {
return start.thenCompose( v -> responseAsyncImpl())
.thenCompose((Response r) -> {
Exchange<T> exch = getExchange();
if (bodyNotPermitted(r)) {
if (bodyIsPresent(r)) {
IOException ioe = new IOException(
"unexpected content length header with 204 response");
exch.cancel(ioe);
return MinimalFuture.failedFuture(ioe);
} else
return handleNoBody(r, exch);
}
return exch.readBodyAsync(responseHandler)
.thenApply((T body) -> {
this.response =
new HttpResponseImpl<>(r.request(), r, this.response, body, exch);
return this.response;
});
});
}
static class NullSubscription implements Flow.Subscription {
@Override
public void request(long n) {
}
@Override
public void cancel() {
}
}
private CompletableFuture<Response> responseAsyncImpl() {
CompletableFuture<Response> cf;
if (attempts.incrementAndGet() > max_attempts) {
cf = failedFuture(new IOException("Too many retries", retryCause));
} else {
if (currentreq.timeout().isPresent()) {
responseTimerEvent = ResponseTimerEvent.of(this);
client.registerTimer(responseTimerEvent);
}
try {
if (currentreq != previousreq) {
requestFilters(currentreq);
}
} catch (IOException e) {
return failedFuture(e);
}
Exchange<T> exch = getExchange();
cf = exch.responseAsync()
.thenCompose((Response response) -> {
HttpRequestImpl newrequest;
try {
newrequest = responseFilters(response);
} catch (IOException e) {
return failedFuture(e);
}
if (newrequest == null) {
if (attempts.get() > 1) {
Log.logError("Succeeded on attempt: " + attempts);
}
return completedFuture(response);
} else {
this.response =
new HttpResponseImpl<>(currentreq, response, this.response, null, exch);
Exchange<T> oldExch = exch;
return exch.ignoreBody().handle((r,t) -> {
previousreq = currentreq;
currentreq = newrequest;
expiredOnce = false;
setExchange(new Exchange<>(currentreq, this, acc));
return responseAsyncImpl();
}).thenCompose(Function.identity());
} })
.handle((response, ex) -> {
cancelTimer();
if (ex == null) {
assert response != null;
return completedFuture(response);
}
CompletableFuture<Response> errorCF = getExceptionalCF(ex);
if (errorCF == null) {
return responseAsyncImpl();
} else {
return errorCF;
} })
.thenCompose(Function.identity());
}
return cf;
}
private static boolean retryPostValue() {
String s = Utils.getNetProperty("jdk.httpclient.enableAllMethodRetry");
if (s == null)
return false;
return s.isEmpty() ? true : Boolean.parseBoolean(s);
}
private static boolean retryConnect() {
String s = Utils.getNetProperty("jdk.httpclient.disableRetryConnect");
if (s == null)
return false;
return s.isEmpty() ? true : Boolean.parseBoolean(s);
}
private static final boolean RETRY_ALWAYS = retryPostValue();
private static final boolean RETRY_CONNECT = retryConnect();
private static boolean isIdempotentRequest(HttpRequest request) {
String method = request.method();
switch (method) {
case "GET" :
case "HEAD" :
return true;
default :
return false;
}
}
private static boolean canRetryRequest(HttpRequest request) {
if (RETRY_ALWAYS)
return true;
if (isIdempotentRequest(request))
return true;
return false;
}
private boolean retryOnFailure(Throwable t) {
return t instanceof ConnectionExpiredException
|| (RETRY_CONNECT && (t instanceof ConnectException));
}
private Throwable retryCause(Throwable t) {
Throwable cause = t instanceof ConnectionExpiredException ? t.getCause() : t;
return cause == null ? t : cause;
}
private CompletableFuture<Response> getExceptionalCF(Throwable t) {
if ((t instanceof CompletionException) || (t instanceof ExecutionException)) {
if (t.getCause() != null) {
t = t.getCause();
}
}
if (cancelled && t instanceof IOException) {
if (!(t instanceof HttpTimeoutException)) {
t = toTimeoutException((IOException)t);
}
} else if (retryOnFailure(t)) {
Throwable cause = retryCause(t);
if (!(t instanceof ConnectException)) {
if (!canRetryRequest(currentreq)) {
return failedFuture(cause);
}
}
retryCause = cause;
if (!expiredOnce) {
if (debug.on())
debug.log(t.getClass().getSimpleName() + " (async): retrying...", t);
expiredOnce = true;
previousreq = currentreq;
return null;
} else {
if (debug.on()) {
debug.log(t.getClass().getSimpleName()
+ " (async): already retried once.", t);
}
t = cause;
}
}
return failedFuture(t);
}
private HttpTimeoutException toTimeoutException(IOException ioe) {
HttpTimeoutException t = null;
Exchange<?> exchange = getExchange();
if (exchange != null) {
ExchangeImpl<?> exchangeImpl = exchange.exchImpl;
if (exchangeImpl != null) {
if (exchangeImpl.connection().connected()) {
t = new HttpTimeoutException("request timed out");
t.initCause(ioe);
}
}
}
if (t == null) {
t = new HttpConnectTimeoutException("HTTP connect timed out");
t.initCause(new ConnectException("HTTP connect timed out"));
}
return t;
}
}