jdk-sandbox: comparison src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/MultiExchange.java

equal deleted inserted replaced

-:e947a3a50a95
+:634d8e14c172
 */
 package jdk.incubator.http;
 import java.io.IOException;
+import java.lang.System.Logger.Level;
 import java.time.Duration;
 import java.util.List;
 import java.security.AccessControlContext;
-import java.security.AccessController;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
-import java.util.function.BiFunction;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.UnaryOperator;
+import jdk.incubator.http.HttpResponse.UntrustedBodyHandler;
 import jdk.incubator.http.internal.common.Log;
 import jdk.incubator.http.internal.common.MinimalFuture;
-import jdk.incubator.http.internal.common.Pair;
+import jdk.incubator.http.internal.common.ConnectionExpiredException;
 import jdk.incubator.http.internal.common.Utils;
-import static jdk.incubator.http.internal.common.Pair.pair;
+import static jdk.incubator.http.internal.common.MinimalFuture.completedFuture;
+import static jdk.incubator.http.internal.common.MinimalFuture.failedFuture;
 /**
 * Encapsulates multiple Exchanges belonging to one HttpRequestImpl.
 * - manages filters
 * - retries due to filters.
 *
 * Creates a new Exchange for each request/response interaction
 */
 class MultiExchange<U,T> {
+static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.
+static final System.Logger DEBUG_LOGGER =
+Utils.getDebugLogger("MultiExchange"::toString, DEBUG);
 private final HttpRequest userRequest; // the user request
 private final HttpRequestImpl request; // a copy of the user request
 final AccessControlContext acc;
 final HttpClientImpl client;
 final HttpResponse.BodyHandler<T> responseHandler;
-final ExecutorWrapper execWrapper;
 final Executor executor;
-final HttpResponse.MultiProcessor<U,T> multiResponseHandler;
+final HttpResponse.MultiSubscriber<U,T> multiResponseSubscriber;
+final AtomicInteger attempts = new AtomicInteger();
 HttpRequestImpl currentreq; // used for async only
 Exchange<T> exchange; // the current exchange
 Exchange<T> previous;
-int attempts;
+volatile Throwable retryCause;
+volatile boolean expiredOnce;
 // Maximum number of times a request will be retried/redirected
 // for any reason
 static final int DEFAULT_MAX_ATTEMPTS = 5;
 static final int max_attempts = Utils.getIntegerNetProperty(
 /**
 * MultiExchange with one final response.
 */
 MultiExchange(HttpRequest req,
 HttpClientImpl client,
-HttpResponse.BodyHandler<T> responseHandler) {
+HttpResponse.BodyHandler<T> responseHandler,
+AccessControlContext acc) {
 this.previous = null;
 this.userRequest = req;
-this.request = new HttpRequestImpl(req);
+this.request = new HttpRequestImpl(req, acc);
 this.currentreq = request;
-this.attempts = 0;
 this.client = client;
 this.filters = client.filterChain();
-if (System.getSecurityManager() != null) {
+this.acc = acc;
-this.acc = AccessController.getContext();
+this.executor = client.theExecutor();
-} else {
-this.acc = null;
-}
-this.execWrapper = new ExecutorWrapper(client.executor(), acc);
-this.executor = execWrapper.executor();
 this.responseHandler = responseHandler;
+if (acc != null) {
+// Restricts the file publisher with the senders ACC, if any
+if (responseHandler instanceof UntrustedBodyHandler)
+((UntrustedBodyHandler)this.responseHandler).setAccessControlContext(acc);
+}
 this.exchange = new Exchange<>(request, this);
-this.multiResponseHandler = null;
+this.multiResponseSubscriber = null;
 this.pushGroup = null;
 }
 /**
 * MultiExchange with multiple responses (HTTP/2 server pushes).
 */
 MultiExchange(HttpRequest req,
 HttpClientImpl client,
-HttpResponse.MultiProcessor<U, T> multiResponseHandler) {
+HttpResponse.MultiSubscriber<U, T> multiResponseSubscriber,
+AccessControlContext acc) {
 this.previous = null;
 this.userRequest = req;
-this.request = new HttpRequestImpl(req);
+this.request = new HttpRequestImpl(req, acc);
 this.currentreq = request;
-this.attempts = 0;
 this.client = client;
 this.filters = client.filterChain();
-if (System.getSecurityManager() != null) {
+this.acc = acc;
-this.acc = AccessController.getContext();
+this.executor = client.theExecutor();
-} else {
+this.multiResponseSubscriber = multiResponseSubscriber;
-this.acc = null;
+this.pushGroup = new PushGroup<>(multiResponseSubscriber, request, acc);
-}
-this.execWrapper = new ExecutorWrapper(client.executor(), acc);
-this.executor = execWrapper.executor();
-this.multiResponseHandler = multiResponseHandler;
-this.pushGroup = new PushGroup<>(multiResponseHandler, request);
 this.exchange = new Exchange<>(request, this);
 this.responseHandler = pushGroup.mainResponseHandler();
 }
-public HttpResponseImpl<T> response() throws IOException, InterruptedException {
-HttpRequestImpl r = request;
-if (r.duration() != null) {
-timedEvent = new TimedEvent(r.duration());
-client.registerTimer(timedEvent);
-}
-while (attempts < max_attempts) {
-try {
-attempts++;
-Exchange<T> currExchange = getExchange();
-requestFilters(r);
-Response response = currExchange.response();
-HttpRequestImpl newreq = responseFilters(response);
-if (newreq == null) {
-if (attempts > 1) {
-Log.logError("Succeeded on attempt: " + attempts);
-}
-T body = currExchange.readBody(responseHandler);
-cancelTimer();
-return new HttpResponseImpl<>(userRequest, response, body, currExchange);
-}
-//response.body(HttpResponse.ignoreBody());
-setExchange(new Exchange<>(newreq, this, acc));
-r = newreq;
-} catch (IOException e) {
-if (cancelled) {
-throw new HttpTimeoutException("Request timed out");
-}
-throw e;
-}
-}
-cancelTimer();
-throw new IOException("Retry limit exceeded");
-}
 CompletableFuture<Void> multiCompletionCF() {
 return pushGroup.groupResult();
 }
 private synchronized Exchange<T> getExchange() {
 HttpClient.Version version() {
 return request.version().orElse(client.version());
 }
 private synchronized void setExchange(Exchange<T> exchange) {
+if (this.exchange != null && exchange != this.exchange) {
+this.exchange.released();
+}
 this.exchange = exchange;
 }
 private void cancelTimer() {
 if (timedEvent != null) {
 public void cancel(IOException cause) {
 cancelled = true;
 getExchange().cancel(cause);
 }
-public CompletableFuture<HttpResponseImpl<T>> responseAsync() {
+public CompletableFuture<HttpResponse<T>> responseAsync() {
 CompletableFuture<Void> start = new MinimalFuture<>();
-CompletableFuture<HttpResponseImpl<T>> cf = responseAsync0(start);
+CompletableFuture<HttpResponse<T>> cf = responseAsync0(start);
 start.completeAsync( () -> null, executor); // trigger execution
 return cf;
 }
-private CompletableFuture<HttpResponseImpl<T>> responseAsync0(CompletableFuture<Void> start) {
+private CompletableFuture<HttpResponse<T>>
+responseAsync0(CompletableFuture<Void> start) {
 return start.thenCompose( v -> responseAsyncImpl())
 .thenCompose((Response r) -> {
 Exchange<T> exch = getExchange();
 return exch.readBodyAsync(responseHandler)
-.thenApply((T body) ->  new HttpResponseImpl<>(userRequest, r, body, exch));
+.thenApply((T body) ->
-});
+new HttpResponseImpl<>(userRequest,
+r,
+body,
+exch));
+});
 }
 CompletableFuture<U> multiResponseAsync() {
 CompletableFuture<Void> start = new MinimalFuture<>();
-CompletableFuture<HttpResponseImpl<T>> cf = responseAsync0(start);
+CompletableFuture<HttpResponse<T>> cf = responseAsync0(start);
 CompletableFuture<HttpResponse<T>> mainResponse =
-cf.thenApply((HttpResponseImpl<T> b) -> {
+cf.thenApply(b -> {
-multiResponseHandler.onResponse(b);
+multiResponseSubscriber.onResponse(b);
-return (HttpResponse<T>)b;
+pushGroup.noMorePushes(true);
-});
+return b; });
 pushGroup.setMainResponse(mainResponse);
-// set up house-keeping related to multi-response
+CompletableFuture<U> res = multiResponseSubscriber.completion(pushGroup.groupResult(),
-mainResponse.thenAccept((r) -> {
+pushGroup.pushesCF());
-// All push promises received by now.
-pushGroup.noMorePushes(true);
-});
-CompletableFuture<U> res = multiResponseHandler.completion(pushGroup.groupResult(), pushGroup.pushesCF());
 start.completeAsync( () -> null, executor); // trigger execution
 return res;
 }
 private CompletableFuture<Response> responseAsyncImpl() {
 CompletableFuture<Response> cf;
-if (++attempts > max_attempts) {
+if (attempts.incrementAndGet() > max_attempts) {
-cf = MinimalFuture.failedFuture(new IOException("Too many retries"));
+cf = failedFuture(new IOException("Too many retries", retryCause));
 } else {
-if (currentreq.duration() != null) {
+if (currentreq.timeout().isPresent()) {
-timedEvent = new TimedEvent(currentreq.duration());
+timedEvent = new TimedEvent(currentreq.timeout().get());
 client.registerTimer(timedEvent);
 }
 try {
-// 1. Apply request filters
+// 1. apply request filters
 requestFilters(currentreq);
 } catch (IOException e) {
-return MinimalFuture.failedFuture(e);
+return failedFuture(e);
 }
 Exchange<T> exch = getExchange();
 // 2. get response
 cf = exch.responseAsync()
 .thenCompose((Response response) -> {
-HttpRequestImpl newrequest = null;
+HttpRequestImpl newrequest;
 try {
-// 3. Apply response filters
+// 3. apply response filters
 newrequest = responseFilters(response);
 } catch (IOException e) {
-return MinimalFuture.failedFuture(e);
+return failedFuture(e);
-}
-// 4. Check filter result and repeat or continue
-if (newrequest == null) {
-if (attempts > 1) {
-Log.logError("Succeeded on attempt: " + attempts);
 }
-return MinimalFuture.completedFuture(response);
+// 4. check filter result and repeat or continue
-} else {
+if (newrequest == null) {
-currentreq = newrequest;
+if (attempts.get() > 1) {
-setExchange(new Exchange<>(currentreq, this, acc));
+Log.logError("Succeeded on attempt: " + attempts);
-//reads body off previous, and then waits for next response
+}
-return responseAsyncImpl();
+return completedFuture(response);
-}
+} else {
-})
+currentreq = newrequest;
-// 5. Handle errors and cancel any timer set
+expiredOnce = false;
-.handle((response, ex) -> {
+setExchange(new Exchange<>(currentreq, this, acc));
-cancelTimer();
+//reads body off previous, and then waits for next response
-if (ex == null) {
+return responseAsyncImpl();
-assert response != null;
+} })
-return MinimalFuture.completedFuture(response);
+.handle((response, ex) -> {
-}
+// 5. handle errors and cancel any timer set
-// all exceptions thrown are handled here
+cancelTimer();
-CompletableFuture<Response> error = getExceptionalCF(ex);
+if (ex == null) {
-if (error == null) {
+assert response != null;
-return responseAsyncImpl();
+return completedFuture(response);
-} else {
+}
-return error;
+// all exceptions thrown are handled here
-}
+CompletableFuture<Response> errorCF = getExceptionalCF(ex);
-})
+if (errorCF == null) {
-.thenCompose(UnaryOperator.identity());
+return responseAsyncImpl();
+} else {
+return errorCF;
+} })
+.thenCompose(UnaryOperator.identity());
 }
 return cf;
 }
 /**
-* Take a Throwable and return a suitable CompletableFuture that is
+* Takes a Throwable and returns a suitable CompletableFuture that is
-* completed exceptionally.
+* completed exceptionally, or null.
 */
 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) {
 t = new HttpTimeoutException("request timed out");
-}
+} else if (t instanceof ConnectionExpiredException) {
-return MinimalFuture.failedFuture(t);
+// allow the retry mechanism to do its work
+// ####: method (GET,HEAD, not POST?), no bytes written or read ( differentiate? )
+if (t.getCause() != null) retryCause = t.getCause();
+if (!expiredOnce) {
+DEBUG_LOGGER.log(Level.DEBUG,
+"MultiExchange: ConnectionExpiredException (async): retrying...",
+t);
+expiredOnce = true;
+return null;
+} else {
+DEBUG_LOGGER.log(Level.DEBUG,
+"MultiExchange: ConnectionExpiredException (async): already retried once.",
+t);
+if (t.getCause() != null) t = t.getCause();
+}
+}
+return failedFuture(t);
 }
 class TimedEvent extends TimeoutEvent {
 TimedEvent(Duration duration) {
 super(duration);
 }
 @Override
 public void handle() {
+DEBUG_LOGGER.log(Level.DEBUG,
+"Cancelling MultiExchange due to timeout for request %s",
+request);
 cancel(new HttpTimeoutException("request timed out"));
 }
 }
 }

branch	http-client-branch
changeset 55763	634d8e14c172
parent 47216	71c04702a3d5
child 55764	34d7cc00f87a