8245462: HttpClient send throws InterruptedException when interrupted but does not cancel request
Allows an HTTP operation to be cancelled by calling CompletableFuture::cancel(true) Reviewed-by: michaelm, chegar, alanb
This commit is contained in:
Daniel Fuchs
parent
13918a4519
commit
80d889189a
src/java.net.http/share/classes
java/net/http
jdk/internal/net/http
test/jdk/java/net/httpclient
@ -40,8 +40,6 @@ import java.time.Duration;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import javax.net.ssl.SSLContext;
|
||||
import javax.net.ssl.SSLParameters;
|
||||
import java.net.http.HttpResponse.BodyHandler;
|
||||
@ -53,8 +51,10 @@ import jdk.internal.net.http.HttpClientBuilderImpl;
|
||||
*
|
||||
* <p> An {@code HttpClient} can be used to send {@linkplain HttpRequest
|
||||
* requests} and retrieve their {@linkplain HttpResponse responses}. An {@code
|
||||
* HttpClient} is created through a {@link HttpClient#newBuilder() builder}. The
|
||||
* builder can be used to configure per-client state, like: the preferred
|
||||
* HttpClient} is created through a {@link HttpClient.Builder builder}.
|
||||
* The {@link #newBuilder() newBuilder} method returns a builder that creates
|
||||
* instances of the default {@code HttpClient} implementation.
|
||||
* The builder can be used to configure per-client state, like: the preferred
|
||||
* protocol version ( HTTP/1.1 or HTTP/2 ), whether to follow redirects, a
|
||||
* proxy, an authenticator, etc. Once built, an {@code HttpClient} is immutable,
|
||||
* and can be used to send multiple requests.
|
||||
@ -165,6 +165,9 @@ public abstract class HttpClient {
|
||||
/**
|
||||
* Creates a new {@code HttpClient} builder.
|
||||
*
|
||||
* <p> Builders returned by this method create instances
|
||||
* of the default {@code HttpClient} implementation.
|
||||
*
|
||||
* @return an {@code HttpClient.Builder}
|
||||
*/
|
||||
public static Builder newBuilder() {
|
||||
@ -529,6 +532,23 @@ public abstract class HttpClient {
|
||||
* response status, headers, and body ( as handled by given response body
|
||||
* handler ).
|
||||
*
|
||||
* <p> If the operation is interrupted, the default {@code HttpClient}
|
||||
* implementation attempts to cancel the HTTP exchange and
|
||||
* {@link InterruptedException} is thrown.
|
||||
* No guarantee is made as to exactly <em>when</em> the cancellation request
|
||||
* may be taken into account. In particular, the request might still get sent
|
||||
* to the server, as its processing might already have started asynchronously
|
||||
* in another thread, and the underlying resources may only be released
|
||||
* asynchronously.
|
||||
* <ul>
|
||||
* <li>With HTTP/1.1, an attempt to cancel may cause the underlying
|
||||
* connection to be closed abruptly.
|
||||
* <li>With HTTP/2, an attempt to cancel may cause the stream to be reset,
|
||||
* or in certain circumstances, may also cause the connection to be
|
||||
* closed abruptly, if, for instance, the thread is currently trying
|
||||
* to write to the underlying socket.
|
||||
* </ul>
|
||||
*
|
||||
* @param <T> the response body type
|
||||
* @param request the request
|
||||
* @param responseBodyHandler the response body handler
|
||||
@ -588,6 +608,24 @@ public abstract class HttpClient {
|
||||
* information.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p> The default {@code HttpClient} implementation returns
|
||||
* {@code CompletableFuture} objects that are <em>cancelable</em>.
|
||||
* {@code CompletableFuture} objects {@linkplain CompletableFuture#newIncompleteFuture()
|
||||
* derived} from cancelable futures are themselves <em>cancelable</em>.
|
||||
* Invoking {@linkplain CompletableFuture#cancel(boolean) cancel(true)}
|
||||
* on a cancelable future that is not completed, attempts to cancel the HTTP exchange
|
||||
* in an effort to release underlying resources as soon as possible.
|
||||
* No guarantee is made as to exactly <em>when</em> the cancellation request
|
||||
* may be taken into account. In particular, the request might still get sent
|
||||
* to the server, as its processing might already have started asynchronously
|
||||
* in another thread, and the underlying resources may only be released
|
||||
* asynchronously.
|
||||
* <ul>
|
||||
* <li>With HTTP/1.1, an attempt to cancel may cause the underlying connection
|
||||
* to be closed abruptly.
|
||||
* <li>With HTTP/2, an attempt to cancel may cause the stream to be reset.
|
||||
* </ul>
|
||||
*
|
||||
* @param <T> the response body type
|
||||
* @param request the request
|
||||
* @param responseBodyHandler the response body handler
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2015, 2020, 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
|
||||
@ -81,7 +81,8 @@ final class Exchange<T> {
|
||||
final AccessControlContext acc;
|
||||
final MultiExchange<T> multi;
|
||||
final Executor parentExecutor;
|
||||
boolean upgrading; // to HTTP/2
|
||||
volatile boolean upgrading; // to HTTP/2
|
||||
volatile boolean upgraded; // to HTTP/2
|
||||
final PushGroup<T> pushGroup;
|
||||
final String dbgTag;
|
||||
|
||||
@ -139,12 +140,15 @@ final class Exchange<T> {
|
||||
// exchange so that it can be aborted/timed out mid setup.
|
||||
static final class ConnectionAborter {
|
||||
private volatile HttpConnection connection;
|
||||
private volatile boolean closeRequested;
|
||||
|
||||
void connection(HttpConnection connection) {
|
||||
this.connection = connection;
|
||||
if (closeRequested) closeConnection();
|
||||
}
|
||||
|
||||
void closeConnection() {
|
||||
closeRequested = true;
|
||||
HttpConnection connection = this.connection;
|
||||
this.connection = null;
|
||||
if (connection != null) {
|
||||
@ -155,6 +159,11 @@ final class Exchange<T> {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void disable() {
|
||||
connection = null;
|
||||
closeRequested = false;
|
||||
}
|
||||
}
|
||||
|
||||
// Called for 204 response - when no body is permitted
|
||||
@ -278,6 +287,30 @@ final class Exchange<T> {
|
||||
}
|
||||
}
|
||||
|
||||
<T> CompletableFuture<T> checkCancelled(CompletableFuture<T> 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();
|
||||
} catch (Throwable x) {
|
||||
if (debug.on()) debug.log("Failed to close connection", x);
|
||||
}
|
||||
return MinimalFuture.<T>failedFuture(t);
|
||||
}
|
||||
}
|
||||
}
|
||||
return cf;
|
||||
}).thenCompose(Function.identity());
|
||||
}
|
||||
|
||||
public void h2Upgrade() {
|
||||
upgrading = true;
|
||||
request.setH2Upgrade(client.client2());
|
||||
@ -299,7 +332,10 @@ final class Exchange<T> {
|
||||
Throwable t = getCancelCause();
|
||||
checkCancelled();
|
||||
if (t != null) {
|
||||
return MinimalFuture.failedFuture(t);
|
||||
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;
|
||||
@ -481,11 +517,14 @@ final class Exchange<T> {
|
||||
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) -> {
|
||||
boolean cached = c.offerConnection();
|
||||
if (cached) connectionAborter.disable();
|
||||
Stream<T> s = c.getStream(1);
|
||||
|
||||
if (s == null) {
|
||||
@ -520,11 +559,12 @@ final class Exchange<T> {
|
||||
}
|
||||
// Check whether the HTTP/1.1 was cancelled.
|
||||
if (t == null) t = e.getCancelCause();
|
||||
// if HTTP/1.1 exchange was timed out, don't
|
||||
// try to go further.
|
||||
if (t instanceof HttpTimeoutException) {
|
||||
s.cancelImpl(t);
|
||||
return MinimalFuture.failedFuture(t);
|
||||
// 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);
|
||||
|
||||
@ -228,4 +228,9 @@ abstract class ExchangeImpl<T> {
|
||||
* @return the cause for which this exchange was canceled, if available.
|
||||
*/
|
||||
abstract Throwable getCancelCause();
|
||||
|
||||
// Mark the exchange as upgraded
|
||||
// Needed to handle cancellation during the upgrade from
|
||||
// Http1Exchange to Stream
|
||||
void upgraded() { }
|
||||
}
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2015, 2020, 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
|
||||
@ -62,6 +62,7 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
final HttpClientImpl client;
|
||||
final Executor executor;
|
||||
private final Http1AsyncReceiver asyncReceiver;
|
||||
private volatile boolean upgraded;
|
||||
|
||||
/** Records a possible cancellation raised before any operation
|
||||
* has been initiated, or an error received while sending the request. */
|
||||
@ -487,10 +488,15 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
connection.close();
|
||||
if (!upgraded)
|
||||
connection.close();
|
||||
}
|
||||
}
|
||||
|
||||
void upgraded() {
|
||||
upgraded = true;
|
||||
}
|
||||
|
||||
private void runInline(Runnable run) {
|
||||
assert !client.isSelectorThread();
|
||||
run.run();
|
||||
@ -543,6 +549,16 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
}
|
||||
}
|
||||
|
||||
private void cancelUpstreamSubscription() {
|
||||
final Executor exec = client.theExecutor();
|
||||
if (debug.on()) debug.log("cancelling upstream publisher");
|
||||
if (bodySubscriber != null) {
|
||||
exec.execute(bodySubscriber::cancelSubscription);
|
||||
} else if (debug.on()) {
|
||||
debug.log("bodySubscriber is null");
|
||||
}
|
||||
}
|
||||
|
||||
// Invoked only by the publisher
|
||||
// ALL tasks should execute off the Selector-Manager thread
|
||||
/** Returns the next portion of the HTTP request, or the error. */
|
||||
@ -551,12 +567,7 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
final DataPair dp = outgoing.pollFirst();
|
||||
|
||||
if (writePublisher.cancelled) {
|
||||
if (debug.on()) debug.log("cancelling upstream publisher");
|
||||
if (bodySubscriber != null) {
|
||||
exec.execute(bodySubscriber::cancelSubscription);
|
||||
} else if (debug.on()) {
|
||||
debug.log("bodySubscriber is null");
|
||||
}
|
||||
cancelUpstreamSubscription();
|
||||
headersSentCF.completeAsync(() -> this, exec);
|
||||
bodySentCF.completeAsync(() -> this, exec);
|
||||
return null;
|
||||
@ -642,6 +653,30 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
return tag;
|
||||
}
|
||||
|
||||
@SuppressWarnings("fallthrough")
|
||||
private boolean checkRequestCancelled() {
|
||||
if (exchange.multi.requestCancelled()) {
|
||||
if (debug.on()) debug.log("request cancelled");
|
||||
if (subscriber == null) {
|
||||
if (debug.on()) debug.log("no subscriber yet");
|
||||
return true;
|
||||
}
|
||||
switch (state) {
|
||||
case BODY:
|
||||
cancelUpstreamSubscription();
|
||||
// fall trough to HEADERS
|
||||
case HEADERS:
|
||||
Throwable cause = getCancelCause();
|
||||
if (cause == null) cause = new IOException("Request cancelled");
|
||||
subscriber.onError(cause);
|
||||
writeScheduler.stop();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
final class WriteTask implements Runnable {
|
||||
@Override
|
||||
public void run() {
|
||||
@ -655,10 +690,13 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
return;
|
||||
}
|
||||
|
||||
if (checkRequestCancelled()) return;
|
||||
|
||||
if (subscriber == null) {
|
||||
if (debug.on()) debug.log("no subscriber yet");
|
||||
return;
|
||||
}
|
||||
|
||||
if (debug.on()) debug.log(() -> "hasOutgoing = " + hasOutgoing());
|
||||
while (hasOutgoing() && demand.tryDecrement()) {
|
||||
DataPair dp = getOutgoing();
|
||||
@ -683,6 +721,7 @@ class Http1Exchange<T> extends ExchangeImpl<T> {
|
||||
// The next Subscriber will eventually take over.
|
||||
|
||||
} else {
|
||||
if (checkRequestCancelled()) return;
|
||||
if (debug.on())
|
||||
debug.log("onNext with " + Utils.remaining(data) + " bytes");
|
||||
subscriber.onNext(data);
|
||||
|
||||
@ -526,6 +526,10 @@ final class HttpClientImpl extends HttpClient implements Trackable {
|
||||
throws IOException, InterruptedException
|
||||
{
|
||||
CompletableFuture<HttpResponse<T>> cf = null;
|
||||
|
||||
// if the thread is already interrupted no need to go further.
|
||||
// cf.get() would throw anyway.
|
||||
if (Thread.interrupted()) throw new InterruptedException();
|
||||
try {
|
||||
cf = sendAsync(req, responseHandler, null, null);
|
||||
return cf.get();
|
||||
|
||||
@ -26,6 +26,7 @@
|
||||
package jdk.internal.net.http;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.lang.ref.WeakReference;
|
||||
import java.net.ConnectException;
|
||||
import java.net.http.HttpConnectTimeoutException;
|
||||
import java.time.Duration;
|
||||
@ -34,6 +35,7 @@ import java.util.LinkedList;
|
||||
import java.security.AccessControlContext;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.CancellationException;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.CompletionException;
|
||||
@ -42,6 +44,7 @@ import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Flow;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.Function;
|
||||
|
||||
import java.net.http.HttpClient;
|
||||
@ -51,6 +54,7 @@ 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.Cancelable;
|
||||
import jdk.internal.net.http.common.Log;
|
||||
import jdk.internal.net.http.common.Logger;
|
||||
import jdk.internal.net.http.common.MinimalFuture;
|
||||
@ -67,7 +71,7 @@ import static jdk.internal.net.http.common.MinimalFuture.failedFuture;
|
||||
*
|
||||
* Creates a new Exchange for each request/response interaction
|
||||
*/
|
||||
class MultiExchange<T> {
|
||||
class MultiExchange<T> implements Cancelable {
|
||||
|
||||
static final Logger debug =
|
||||
Utils.getDebugLogger("MultiExchange"::toString, Utils.DEBUG);
|
||||
@ -90,7 +94,6 @@ class MultiExchange<T> {
|
||||
|
||||
// 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(
|
||||
"jdk.httpclient.redirects.retrylimit", DEFAULT_MAX_ATTEMPTS
|
||||
@ -99,6 +102,7 @@ class MultiExchange<T> {
|
||||
private final LinkedList<HeaderFilter> filters;
|
||||
ResponseTimerEvent responseTimerEvent;
|
||||
volatile boolean cancelled;
|
||||
AtomicReference<CancellationException> interrupted = new AtomicReference<>();
|
||||
final PushGroup<T> pushGroup;
|
||||
|
||||
/**
|
||||
@ -176,6 +180,20 @@ class MultiExchange<T> {
|
||||
this.exchange = new Exchange<>(request, this);
|
||||
}
|
||||
|
||||
static final class CancelableRef implements Cancelable {
|
||||
private final WeakReference<Cancelable> cancelableRef;
|
||||
CancelableRef(Cancelable cancelable) {
|
||||
cancelableRef = new WeakReference<>(cancelable);
|
||||
}
|
||||
@Override
|
||||
public boolean cancel(boolean mayInterruptIfRunning) {
|
||||
Cancelable cancelable = cancelableRef.get();
|
||||
if (cancelable != null) {
|
||||
return cancelable.cancel(mayInterruptIfRunning);
|
||||
} else return false;
|
||||
}
|
||||
}
|
||||
|
||||
synchronized Exchange<T> getExchange() {
|
||||
return exchange;
|
||||
}
|
||||
@ -194,6 +212,7 @@ class MultiExchange<T> {
|
||||
private synchronized void setExchange(Exchange<T> exchange) {
|
||||
if (this.exchange != null && exchange != this.exchange) {
|
||||
this.exchange.released();
|
||||
if (cancelled) exchange.cancel();
|
||||
}
|
||||
this.exchange = exchange;
|
||||
}
|
||||
@ -240,8 +259,36 @@ class MultiExchange<T> {
|
||||
getExchange().cancel(cause);
|
||||
}
|
||||
|
||||
/**
|
||||
* Used to relay a call from {@link CompletableFuture#cancel(boolean)}
|
||||
* to this multi exchange for the purpose of cancelling the
|
||||
* HTTP exchange.
|
||||
* @param mayInterruptIfRunning if true, and this exchange is not already
|
||||
* cancelled, this method will attempt to interrupt and cancel the
|
||||
* exchange. Otherwise, the exchange is allowed to proceed and this
|
||||
* method does nothing.
|
||||
* @return true if the exchange was cancelled, false otherwise.
|
||||
*/
|
||||
@Override
|
||||
public boolean cancel(boolean mayInterruptIfRunning) {
|
||||
boolean cancelled = this.cancelled;
|
||||
if (!cancelled && mayInterruptIfRunning) {
|
||||
if (interrupted.get() == null) {
|
||||
interrupted.compareAndSet(null,
|
||||
new CancellationException("Request cancelled"));
|
||||
}
|
||||
this.cancelled = true;
|
||||
var exchange = getExchange();
|
||||
if (exchange != null) {
|
||||
exchange.cancel();
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public CompletableFuture<HttpResponse<T>> responseAsync(Executor executor) {
|
||||
CompletableFuture<Void> start = new MinimalFuture<>();
|
||||
CompletableFuture<Void> start = new MinimalFuture<>(new CancelableRef(this));
|
||||
CompletableFuture<HttpResponse<T>> cf = responseAsync0(start);
|
||||
start.completeAsync( () -> null, executor); // trigger execution
|
||||
return cf;
|
||||
@ -308,7 +355,20 @@ class MultiExchange<T> {
|
||||
new HttpResponseImpl<>(r.request(), r, this.response, body, exch);
|
||||
return this.response;
|
||||
});
|
||||
});
|
||||
}).exceptionallyCompose(this::whenCancelled);
|
||||
}
|
||||
|
||||
private CompletableFuture<HttpResponse<T>> whenCancelled(Throwable t) {
|
||||
CancellationException x = interrupted.get();
|
||||
if (x != null) {
|
||||
// make sure to fail with CancellationException if cancel(true)
|
||||
// was called.
|
||||
t = x.initCause(Utils.getCancelCause(t));
|
||||
if (debug.on()) {
|
||||
debug.log("MultiExchange interrupted with: " + t.getCause());
|
||||
}
|
||||
}
|
||||
return MinimalFuture.failedFuture(t);
|
||||
}
|
||||
|
||||
static class NullSubscription implements Flow.Subscription {
|
||||
@ -434,7 +494,17 @@ class MultiExchange<T> {
|
||||
return false;
|
||||
}
|
||||
|
||||
// Returns true if cancel(true) was called.
|
||||
// This is an important distinction in several scenarios:
|
||||
// for instance, if cancel(true) was called 1. we don't want
|
||||
// to retry, 2. we don't want to wrap the exception in
|
||||
// a timeout exception.
|
||||
boolean requestCancelled() {
|
||||
return interrupted.get() != null;
|
||||
}
|
||||
|
||||
private boolean retryOnFailure(Throwable t) {
|
||||
if (requestCancelled()) return false;
|
||||
return t instanceof ConnectionExpiredException
|
||||
|| (RETRY_CONNECT && (t instanceof ConnectException));
|
||||
}
|
||||
@ -454,7 +524,7 @@ class MultiExchange<T> {
|
||||
t = t.getCause();
|
||||
}
|
||||
}
|
||||
if (cancelled && t instanceof IOException) {
|
||||
if (cancelled && !requestCancelled() && t instanceof IOException) {
|
||||
if (!(t instanceof HttpTimeoutException)) {
|
||||
t = toTimeoutException((IOException)t);
|
||||
}
|
||||
|
||||
@ -103,9 +103,11 @@ class PlainHttpConnection extends HttpConnection {
|
||||
|
||||
final class ConnectEvent extends AsyncEvent {
|
||||
private final CompletableFuture<Void> cf;
|
||||
private final Exchange<?> exchange;
|
||||
|
||||
ConnectEvent(CompletableFuture<Void> cf) {
|
||||
ConnectEvent(CompletableFuture<Void> cf, Exchange<?> exchange) {
|
||||
this.cf = cf;
|
||||
this.exchange = exchange;
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -126,10 +128,10 @@ class PlainHttpConnection extends HttpConnection {
|
||||
if (debug.on())
|
||||
debug.log("ConnectEvent: finishing connect");
|
||||
boolean finished = chan.finishConnect();
|
||||
assert finished : "Expected channel to be connected";
|
||||
if (debug.on())
|
||||
debug.log("ConnectEvent: connect finished: %s Local addr: %s",
|
||||
finished, chan.getLocalAddress());
|
||||
debug.log("ConnectEvent: connect finished: %s, cancelled: %s, Local addr: %s",
|
||||
finished, exchange.multi.requestCancelled(), chan.getLocalAddress());
|
||||
assert finished || exchange.multi.requestCancelled() : "Expected channel to be connected";
|
||||
// complete async since the event runs on the SelectorManager thread
|
||||
cf.completeAsync(() -> null, client().theExecutor());
|
||||
} catch (Throwable e) {
|
||||
@ -173,8 +175,9 @@ class PlainHttpConnection extends HttpConnection {
|
||||
cf.complete(null);
|
||||
} else {
|
||||
if (debug.on()) debug.log("registering connect event");
|
||||
client().registerEvent(new ConnectEvent(cf));
|
||||
client().registerEvent(new ConnectEvent(cf, exchange));
|
||||
}
|
||||
cf = exchange.checkCancelled(cf, this);
|
||||
} catch (Throwable throwable) {
|
||||
cf.completeExceptionally(Utils.toConnectException(throwable));
|
||||
try {
|
||||
|
||||
@ -178,11 +178,13 @@ public final class RequestPublishers {
|
||||
}
|
||||
|
||||
static long computeLength(Iterable<byte[]> bytes) {
|
||||
long len = 0;
|
||||
for (byte[] b : bytes) {
|
||||
len = Math.addExact(len, (long)b.length);
|
||||
}
|
||||
return len;
|
||||
// Avoid iterating just for the purpose of computing
|
||||
// a length, in case iterating is a costly operation
|
||||
// For HTTP/1.1 it means we will be using chunk encoding
|
||||
// when sending the request body.
|
||||
// For HTTP/2 it means we will not send the optional
|
||||
// Content-length header.
|
||||
return -1;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@ -41,7 +41,6 @@ import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Flow;
|
||||
import java.util.concurrent.Flow.Subscription;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.BiPredicate;
|
||||
import java.net.http.HttpClient;
|
||||
@ -415,6 +414,15 @@ class Stream<T> extends ExchangeImpl<T> {
|
||||
this.windowUpdater = new StreamWindowUpdateSender(connection);
|
||||
}
|
||||
|
||||
private boolean checkRequestCancelled() {
|
||||
if (exchange.multi.requestCancelled()) {
|
||||
if (errorRef.get() == null) cancel();
|
||||
else sendCancelStreamFrame();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Entry point from Http2Connection reader thread.
|
||||
*
|
||||
@ -422,9 +430,9 @@ class Stream<T> extends ExchangeImpl<T> {
|
||||
*/
|
||||
void incoming(Http2Frame frame) throws IOException {
|
||||
if (debug.on()) debug.log("incoming: %s", frame);
|
||||
var cancelled = closed || streamState != 0;
|
||||
var cancelled = checkRequestCancelled() || closed;
|
||||
if ((frame instanceof HeaderFrame)) {
|
||||
HeaderFrame hframe = (HeaderFrame)frame;
|
||||
HeaderFrame hframe = (HeaderFrame) frame;
|
||||
if (hframe.endHeaders()) {
|
||||
Log.logTrace("handling response (streamid={0})", streamid);
|
||||
handleResponse();
|
||||
@ -585,7 +593,7 @@ class Stream<T> extends ExchangeImpl<T> {
|
||||
Log.logRequest("PUSH_PROMISE: " + pushRequest.toString());
|
||||
}
|
||||
PushGroup<T> pushGroup = exchange.getPushGroup();
|
||||
if (pushGroup == null) {
|
||||
if (pushGroup == null || exchange.multi.requestCancelled()) {
|
||||
Log.logTrace("Rejecting push promise stream " + streamid);
|
||||
connection.resetStream(pushStream.streamid, ResetFrame.REFUSED_STREAM);
|
||||
pushStream.close();
|
||||
@ -796,7 +804,7 @@ class Stream<T> extends ExchangeImpl<T> {
|
||||
}
|
||||
|
||||
boolean registerStream(int id, boolean registerIfCancelled) {
|
||||
boolean cancelled = closed;
|
||||
boolean cancelled = closed || exchange.multi.requestCancelled();
|
||||
if (!cancelled || registerIfCancelled) {
|
||||
this.streamid = id;
|
||||
connection.putStream(this, streamid);
|
||||
|
||||
@ -0,0 +1,47 @@
|
||||
/*
|
||||
* Copyright (c) 2020, 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.common;
|
||||
|
||||
/**
|
||||
* A functional interface that allows to request cancellation
|
||||
* of a task - which may or may not have already started.
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface Cancelable {
|
||||
/**
|
||||
* Attempts to cancel execution of a task. This attempt may not
|
||||
* succeed if the task has already completed, has already been cancelled,
|
||||
* or could not be cancelled for some other reason.
|
||||
*
|
||||
* @param mayInterruptIfRunning {@code true} if an attempt to stop the
|
||||
* task should be made even if the task has already started; otherwise,
|
||||
* in-progress tasks are allowed to complete.
|
||||
*
|
||||
* @return {@code false} if the task could not be cancelled,
|
||||
* typically because it has already completed normally;
|
||||
* {@code true} otherwise
|
||||
*/
|
||||
boolean cancel(boolean mayInterruptIfRunning);
|
||||
}
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2016, 2020, 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
|
||||
@ -44,6 +44,7 @@ public final class MinimalFuture<T> extends CompletableFuture<T> {
|
||||
|
||||
private final static AtomicLong TOKENS = new AtomicLong();
|
||||
private final long id;
|
||||
private final Cancelable cancelable;
|
||||
|
||||
public static <U> MinimalFuture<U> completedFuture(U value) {
|
||||
MinimalFuture<U> f = new MinimalFuture<>();
|
||||
@ -70,13 +71,18 @@ public final class MinimalFuture<T> extends CompletableFuture<T> {
|
||||
}
|
||||
|
||||
public MinimalFuture() {
|
||||
this(null);
|
||||
}
|
||||
|
||||
public MinimalFuture(Cancelable cancelable) {
|
||||
super();
|
||||
this.id = TOKENS.incrementAndGet();
|
||||
this.cancelable = cancelable;
|
||||
}
|
||||
|
||||
@Override
|
||||
public <U> MinimalFuture<U> newIncompleteFuture() {
|
||||
return new MinimalFuture<>();
|
||||
return new MinimalFuture<>(cancelable);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -94,8 +100,23 @@ public final class MinimalFuture<T> extends CompletableFuture<T> {
|
||||
return super.toString() + " (id=" + id +")";
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean cancel(boolean mayInterruptIfRunning) {
|
||||
boolean result = false;
|
||||
if (cancelable != null && !isDone()) {
|
||||
result = cancelable.cancel(mayInterruptIfRunning);
|
||||
}
|
||||
return super.cancel(mayInterruptIfRunning) || result;
|
||||
}
|
||||
|
||||
private Cancelable cancelable() {
|
||||
return cancelable;
|
||||
}
|
||||
|
||||
public static <U> MinimalFuture<U> of(CompletionStage<U> stage) {
|
||||
MinimalFuture<U> cf = new MinimalFuture<>();
|
||||
Cancelable cancelable = stage instanceof MinimalFuture
|
||||
? ((MinimalFuture)stage).cancelable() : null;
|
||||
MinimalFuture<U> cf = new MinimalFuture<>(cancelable);
|
||||
stage.whenComplete((r,t) -> complete(cf, r, t));
|
||||
return cf;
|
||||
}
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2015, 2020, 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
|
||||
@ -286,15 +286,25 @@ public final class Utils {
|
||||
}
|
||||
|
||||
public static Throwable getCompletionCause(Throwable x) {
|
||||
if (!(x instanceof CompletionException)
|
||||
&& !(x instanceof ExecutionException)) return x;
|
||||
final Throwable cause = x.getCause();
|
||||
if (cause == null) {
|
||||
Throwable cause = x;
|
||||
while ((cause instanceof CompletionException)
|
||||
|| (cause instanceof ExecutionException)) {
|
||||
cause = cause.getCause();
|
||||
}
|
||||
if (cause == null && cause != x) {
|
||||
throw new InternalError("Unexpected null cause", x);
|
||||
}
|
||||
return cause;
|
||||
}
|
||||
|
||||
public static Throwable getCancelCause(Throwable x) {
|
||||
Throwable cause = getCompletionCause(x);
|
||||
if (cause instanceof ConnectionExpiredException) {
|
||||
cause = cause.getCause();
|
||||
}
|
||||
return cause;
|
||||
}
|
||||
|
||||
public static IOException getIOException(Throwable t) {
|
||||
if (t instanceof IOException) {
|
||||
return (IOException) t;
|
||||
|
||||
628
test/jdk/java/net/httpclient/CancelRequestTest.java
Normal file
628
test/jdk/java/net/httpclient/CancelRequestTest.java
Normal file
@ -0,0 +1,628 @@
|
||||
/*
|
||||
* Copyright (c) 2020, 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.
|
||||
*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
/*
|
||||
* @test
|
||||
* @bug 8245462 8229822
|
||||
* @summary Tests cancelling the request.
|
||||
* @library /test/lib http2/server
|
||||
* @key randomness
|
||||
* @build jdk.test.lib.net.SimpleSSLContext HttpServerAdapters
|
||||
* ReferenceTracker CancelRequestTest
|
||||
* @modules java.base/sun.net.www.http
|
||||
* java.net.http/jdk.internal.net.http.common
|
||||
* java.net.http/jdk.internal.net.http.frame
|
||||
* java.net.http/jdk.internal.net.http.hpack
|
||||
* @run testng/othervm -Djdk.internal.httpclient.debug=true
|
||||
* -Djdk.httpclient.enableAllMethodRetry=true
|
||||
* CancelRequestTest
|
||||
*/
|
||||
// * -Dseed=3582896013206826205L
|
||||
// * -Dseed=5784221742235559231L
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import com.sun.net.httpserver.HttpsConfigurator;
|
||||
import com.sun.net.httpserver.HttpsServer;
|
||||
import jdk.test.lib.RandomFactory;
|
||||
import jdk.test.lib.net.SimpleSSLContext;
|
||||
import org.testng.ITestContext;
|
||||
import org.testng.annotations.AfterClass;
|
||||
import org.testng.annotations.AfterTest;
|
||||
import org.testng.annotations.BeforeMethod;
|
||||
import org.testng.annotations.BeforeTest;
|
||||
import org.testng.annotations.DataProvider;
|
||||
import org.testng.annotations.Test;
|
||||
|
||||
import javax.net.ssl.SSLContext;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpConnectTimeoutException;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.net.http.HttpResponse.BodyHandler;
|
||||
import java.net.http.HttpResponse.BodyHandlers;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.concurrent.CancellationException;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import static java.lang.System.arraycopy;
|
||||
import static java.lang.System.out;
|
||||
import static java.nio.charset.StandardCharsets.UTF_8;
|
||||
import static org.testng.Assert.assertEquals;
|
||||
import static org.testng.Assert.assertTrue;
|
||||
|
||||
public class CancelRequestTest implements HttpServerAdapters {
|
||||
|
||||
private static final Random random = RandomFactory.getRandom();
|
||||
|
||||
SSLContext sslContext;
|
||||
HttpTestServer httpTestServer; // HTTP/1.1 [ 4 servers ]
|
||||
HttpTestServer httpsTestServer; // HTTPS/1.1
|
||||
HttpTestServer http2TestServer; // HTTP/2 ( h2c )
|
||||
HttpTestServer https2TestServer; // HTTP/2 ( h2 )
|
||||
String httpURI;
|
||||
String httpsURI;
|
||||
String http2URI;
|
||||
String https2URI;
|
||||
|
||||
static final long SERVER_LATENCY = 75;
|
||||
static final int MAX_CLIENT_DELAY = 75;
|
||||
static final int ITERATION_COUNT = 3;
|
||||
// a shared executor helps reduce the amount of threads created by the test
|
||||
static final Executor executor = new TestExecutor(Executors.newCachedThreadPool());
|
||||
static final ConcurrentMap<String, Throwable> FAILURES = new ConcurrentHashMap<>();
|
||||
static volatile boolean tasksFailed;
|
||||
static final AtomicLong serverCount = new AtomicLong();
|
||||
static final AtomicLong clientCount = new AtomicLong();
|
||||
static final long start = System.nanoTime();
|
||||
public static String now() {
|
||||
long now = System.nanoTime() - start;
|
||||
long secs = now / 1000_000_000;
|
||||
long mill = (now % 1000_000_000) / 1000_000;
|
||||
long nan = now % 1000_000;
|
||||
return String.format("[%d s, %d ms, %d ns] ", secs, mill, nan);
|
||||
}
|
||||
|
||||
final ReferenceTracker TRACKER = ReferenceTracker.INSTANCE;
|
||||
private volatile HttpClient sharedClient;
|
||||
|
||||
static class TestExecutor implements Executor {
|
||||
final AtomicLong tasks = new AtomicLong();
|
||||
Executor executor;
|
||||
TestExecutor(Executor executor) {
|
||||
this.executor = executor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(Runnable command) {
|
||||
long id = tasks.incrementAndGet();
|
||||
executor.execute(() -> {
|
||||
try {
|
||||
command.run();
|
||||
} catch (Throwable t) {
|
||||
tasksFailed = true;
|
||||
System.out.printf(now() + "Task %s failed: %s%n", id, t);
|
||||
System.err.printf(now() + "Task %s failed: %s%n", id, t);
|
||||
FAILURES.putIfAbsent("Task " + id, t);
|
||||
throw t;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
protected boolean stopAfterFirstFailure() {
|
||||
return Boolean.getBoolean("jdk.internal.httpclient.debug");
|
||||
}
|
||||
|
||||
@BeforeMethod
|
||||
void beforeMethod(ITestContext context) {
|
||||
if (stopAfterFirstFailure() && context.getFailedTests().size() > 0) {
|
||||
throw new RuntimeException("some tests failed");
|
||||
}
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
static final void printFailedTests() {
|
||||
out.println("\n=========================");
|
||||
try {
|
||||
out.printf("%n%sCreated %d servers and %d clients%n",
|
||||
now(), serverCount.get(), clientCount.get());
|
||||
if (FAILURES.isEmpty()) return;
|
||||
out.println("Failed tests: ");
|
||||
FAILURES.entrySet().forEach((e) -> {
|
||||
out.printf("\t%s: %s%n", e.getKey(), e.getValue());
|
||||
e.getValue().printStackTrace(out);
|
||||
});
|
||||
if (tasksFailed) {
|
||||
System.out.println("WARNING: Some tasks failed");
|
||||
}
|
||||
} finally {
|
||||
out.println("\n=========================\n");
|
||||
}
|
||||
}
|
||||
|
||||
private String[] uris() {
|
||||
return new String[] {
|
||||
httpURI,
|
||||
httpsURI,
|
||||
http2URI,
|
||||
https2URI,
|
||||
};
|
||||
}
|
||||
|
||||
@DataProvider(name = "asyncurls")
|
||||
public Object[][] asyncurls() {
|
||||
String[] uris = uris();
|
||||
Object[][] result = new Object[uris.length * 2 * 3][];
|
||||
//Object[][] result = new Object[uris.length][];
|
||||
int i = 0;
|
||||
for (boolean mayInterrupt : List.of(true, false, true)) {
|
||||
for (boolean sameClient : List.of(false, true)) {
|
||||
//if (!sameClient) continue;
|
||||
for (String uri : uris()) {
|
||||
String path = sameClient ? "same" : "new";
|
||||
path = path + (mayInterrupt ? "/interrupt" : "/nointerrupt");
|
||||
result[i++] = new Object[]{uri + path, sameClient, mayInterrupt};
|
||||
}
|
||||
}
|
||||
}
|
||||
assert i == uris.length * 2 * 3;
|
||||
// assert i == uris.length ;
|
||||
return result;
|
||||
}
|
||||
|
||||
@DataProvider(name = "urls")
|
||||
public Object[][] alltests() {
|
||||
String[] uris = uris();
|
||||
Object[][] result = new Object[uris.length * 2][];
|
||||
//Object[][] result = new Object[uris.length][];
|
||||
int i = 0;
|
||||
for (boolean sameClient : List.of(false, true)) {
|
||||
//if (!sameClient) continue;
|
||||
for (String uri : uris()) {
|
||||
String path = sameClient ? "same" : "new";
|
||||
path = path + "/interruptThread";
|
||||
result[i++] = new Object[]{uri + path, sameClient};
|
||||
}
|
||||
}
|
||||
assert i == uris.length * 2;
|
||||
// assert i == uris.length ;
|
||||
return result;
|
||||
}
|
||||
|
||||
private HttpClient makeNewClient() {
|
||||
clientCount.incrementAndGet();
|
||||
return TRACKER.track(HttpClient.newBuilder()
|
||||
.proxy(HttpClient.Builder.NO_PROXY)
|
||||
.executor(executor)
|
||||
.sslContext(sslContext)
|
||||
.build());
|
||||
}
|
||||
|
||||
HttpClient newHttpClient(boolean share) {
|
||||
if (!share) return makeNewClient();
|
||||
HttpClient shared = sharedClient;
|
||||
if (shared != null) return shared;
|
||||
synchronized (this) {
|
||||
shared = sharedClient;
|
||||
if (shared == null) {
|
||||
shared = sharedClient = makeNewClient();
|
||||
}
|
||||
return shared;
|
||||
}
|
||||
}
|
||||
|
||||
final static String BODY = "Some string | that ? can | be split ? several | ways.";
|
||||
|
||||
// should accept SSLHandshakeException because of the connectionAborter
|
||||
// with http/2 and should accept Stream 5 cancelled.
|
||||
// => also examine in what measure we should always
|
||||
// rewrap in "Request Cancelled" when the multi exchange was aborted...
|
||||
private static boolean isCancelled(Throwable t) {
|
||||
while (t instanceof ExecutionException) t = t.getCause();
|
||||
if (t instanceof CancellationException) return true;
|
||||
if (t instanceof IOException) return String.valueOf(t).contains("Request cancelled");
|
||||
out.println("Not a cancellation exception: " + t);
|
||||
t.printStackTrace(out);
|
||||
return false;
|
||||
}
|
||||
|
||||
private static void delay() {
|
||||
int delay = random.nextInt(MAX_CLIENT_DELAY);
|
||||
try {
|
||||
System.out.println("client delay: " + delay);
|
||||
Thread.sleep(delay);
|
||||
} catch (InterruptedException x) {
|
||||
out.println("Unexpected exception: " + x);
|
||||
}
|
||||
}
|
||||
|
||||
@Test(dataProvider = "asyncurls")
|
||||
public void testGetSendAsync(String uri, boolean sameClient, boolean mayInterruptIfRunning)
|
||||
throws Exception {
|
||||
HttpClient client = null;
|
||||
uri = uri + "/get";
|
||||
out.printf("%n%s testGetSendAsync(%s, %b, %b)%n", now(), uri, sameClient, mayInterruptIfRunning);
|
||||
for (int i=0; i< ITERATION_COUNT; i++) {
|
||||
if (!sameClient || client == null)
|
||||
client = newHttpClient(sameClient);
|
||||
|
||||
HttpRequest req = HttpRequest.newBuilder(URI.create(uri))
|
||||
.GET()
|
||||
.build();
|
||||
BodyHandler<String> handler = BodyHandlers.ofString();
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
CompletableFuture<HttpResponse<String>> response = client.sendAsync(req, handler);
|
||||
var cf1 = response.whenComplete((r,t) -> System.out.println(t));
|
||||
CompletableFuture<HttpResponse<String>> cf2 = cf1.whenComplete((r,t) -> latch.countDown());
|
||||
out.println("response: " + response);
|
||||
out.println("cf1: " + cf1);
|
||||
out.println("cf2: " + cf2);
|
||||
delay();
|
||||
cf1.cancel(mayInterruptIfRunning);
|
||||
out.println("response after cancel: " + response);
|
||||
out.println("cf1 after cancel: " + cf1);
|
||||
out.println("cf2 after cancel: " + cf2);
|
||||
try {
|
||||
String body = cf2.get().body();
|
||||
assertEquals(body, Stream.of(BODY.split("\\|")).collect(Collectors.joining()));
|
||||
throw new AssertionError("Expected CancellationException not received");
|
||||
} catch (ExecutionException x) {
|
||||
out.println("Got expected exception: " + x);
|
||||
assertTrue(isCancelled(x));
|
||||
}
|
||||
|
||||
// Cancelling the request may cause an IOException instead...
|
||||
boolean hasCancellationException = false;
|
||||
try {
|
||||
cf1.get();
|
||||
} catch (CancellationException | ExecutionException x) {
|
||||
out.println("Got expected exception: " + x);
|
||||
assertTrue(isCancelled(x));
|
||||
hasCancellationException = x instanceof CancellationException;
|
||||
}
|
||||
|
||||
// because it's cf1 that was cancelled then response might not have
|
||||
// completed yet - so wait for it here...
|
||||
try {
|
||||
String body = response.get().body();
|
||||
assertEquals(body, Stream.of(BODY.split("\\|")).collect(Collectors.joining()));
|
||||
if (mayInterruptIfRunning) {
|
||||
// well actually - this could happen... In which case we'll need to
|
||||
// increase the latency in the server handler...
|
||||
throw new AssertionError("Expected Exception not received");
|
||||
}
|
||||
} catch (ExecutionException x) {
|
||||
assertEquals(response.isDone(), true);
|
||||
Throwable wrapped = x.getCause();
|
||||
assertTrue(CancellationException.class.isAssignableFrom(wrapped.getClass()));
|
||||
Throwable cause = wrapped.getCause();
|
||||
out.println("CancellationException cause: " + x);
|
||||
assertTrue(IOException.class.isAssignableFrom(cause.getClass()));
|
||||
if (cause instanceof HttpConnectTimeoutException) {
|
||||
cause.printStackTrace(out);
|
||||
throw new RuntimeException("Unexpected timeout exception", cause);
|
||||
}
|
||||
if (mayInterruptIfRunning) {
|
||||
out.println("Got expected exception: " + wrapped);
|
||||
out.println("\tcause: " + cause);
|
||||
} else {
|
||||
out.println("Unexpected exception: " + wrapped);
|
||||
wrapped.printStackTrace(out);
|
||||
throw x;
|
||||
}
|
||||
}
|
||||
|
||||
assertEquals(response.isDone(), true);
|
||||
assertEquals(response.isCancelled(), false);
|
||||
assertEquals(cf1.isCancelled(), hasCancellationException);
|
||||
assertEquals(cf2.isDone(), true);
|
||||
assertEquals(cf2.isCancelled(), false);
|
||||
assertEquals(latch.getCount(), 0);
|
||||
}
|
||||
}
|
||||
|
||||
@Test(dataProvider = "asyncurls")
|
||||
public void testPostSendAsync(String uri, boolean sameClient, boolean mayInterruptIfRunning)
|
||||
throws Exception {
|
||||
uri = uri + "/post";
|
||||
HttpClient client = null;
|
||||
out.printf("%n%s testPostSendAsync(%s, %b, %b)%n", now(), uri, sameClient, mayInterruptIfRunning);
|
||||
for (int i=0; i< ITERATION_COUNT; i++) {
|
||||
if (!sameClient || client == null)
|
||||
client = newHttpClient(sameClient);
|
||||
|
||||
CompletableFuture<CompletableFuture<?>> cancelFuture = new CompletableFuture<>();
|
||||
|
||||
Iterable<byte[]> iterable = new Iterable<byte[]>() {
|
||||
@Override
|
||||
public Iterator<byte[]> iterator() {
|
||||
// this is dangerous
|
||||
out.println("waiting for completion on: " + cancelFuture);
|
||||
boolean async = random.nextBoolean();
|
||||
Runnable cancel = () -> {
|
||||
out.println("Cancelling from " + Thread.currentThread());
|
||||
var cf1 = cancelFuture.join();
|
||||
cf1.cancel(mayInterruptIfRunning);
|
||||
out.println("cancelled " + cf1);
|
||||
};
|
||||
if (async) executor.execute(cancel);
|
||||
else cancel.run();
|
||||
return List.of(BODY.getBytes(UTF_8)).iterator();
|
||||
}
|
||||
};
|
||||
|
||||
HttpRequest req = HttpRequest.newBuilder(URI.create(uri))
|
||||
.POST(HttpRequest.BodyPublishers.ofByteArrays(iterable))
|
||||
.build();
|
||||
BodyHandler<String> handler = BodyHandlers.ofString();
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
CompletableFuture<HttpResponse<String>> response = client.sendAsync(req, handler);
|
||||
var cf1 = response.whenComplete((r,t) -> System.out.println(t));
|
||||
CompletableFuture<HttpResponse<String>> cf2 = cf1.whenComplete((r,t) -> latch.countDown());
|
||||
out.println("response: " + response);
|
||||
out.println("cf1: " + cf1);
|
||||
out.println("cf2: " + cf2);
|
||||
cancelFuture.complete(cf1);
|
||||
out.println("response after cancel: " + response);
|
||||
out.println("cf1 after cancel: " + cf1);
|
||||
out.println("cf2 after cancel: " + cf2);
|
||||
try {
|
||||
String body = cf2.get().body();
|
||||
assertEquals(body, Stream.of(BODY.split("\\|")).collect(Collectors.joining()));
|
||||
throw new AssertionError("Expected CancellationException not received");
|
||||
} catch (ExecutionException x) {
|
||||
out.println("Got expected exception: " + x);
|
||||
assertTrue(isCancelled(x));
|
||||
}
|
||||
|
||||
// Cancelling the request may cause an IOException instead...
|
||||
boolean hasCancellationException = false;
|
||||
try {
|
||||
cf1.get();
|
||||
} catch (CancellationException | ExecutionException x) {
|
||||
out.println("Got expected exception: " + x);
|
||||
assertTrue(isCancelled(x));
|
||||
hasCancellationException = x instanceof CancellationException;
|
||||
}
|
||||
|
||||
// because it's cf1 that was cancelled then response might not have
|
||||
// completed yet - so wait for it here...
|
||||
try {
|
||||
String body = response.get().body();
|
||||
assertEquals(body, Stream.of(BODY.split("\\|")).collect(Collectors.joining()));
|
||||
if (mayInterruptIfRunning) {
|
||||
// well actually - this could happen... In which case we'll need to
|
||||
// increase the latency in the server handler...
|
||||
throw new AssertionError("Expected Exception not received");
|
||||
}
|
||||
} catch (ExecutionException x) {
|
||||
assertEquals(response.isDone(), true);
|
||||
Throwable wrapped = x.getCause();
|
||||
assertTrue(CancellationException.class.isAssignableFrom(wrapped.getClass()));
|
||||
Throwable cause = wrapped.getCause();
|
||||
assertTrue(IOException.class.isAssignableFrom(cause.getClass()));
|
||||
if (cause instanceof HttpConnectTimeoutException) {
|
||||
cause.printStackTrace(out);
|
||||
throw new RuntimeException("Unexpected timeout exception", cause);
|
||||
}
|
||||
if (mayInterruptIfRunning) {
|
||||
out.println("Got expected exception: " + wrapped);
|
||||
out.println("\tcause: " + cause);
|
||||
} else {
|
||||
out.println("Unexpected exception: " + wrapped);
|
||||
wrapped.printStackTrace(out);
|
||||
throw x;
|
||||
}
|
||||
}
|
||||
|
||||
assertEquals(response.isDone(), true);
|
||||
assertEquals(response.isCancelled(), false);
|
||||
assertEquals(cf1.isCancelled(), hasCancellationException);
|
||||
assertEquals(cf2.isDone(), true);
|
||||
assertEquals(cf2.isCancelled(), false);
|
||||
assertEquals(latch.getCount(), 0);
|
||||
}
|
||||
}
|
||||
|
||||
@Test(dataProvider = "urls")
|
||||
public void testPostInterrupt(String uri, boolean sameClient)
|
||||
throws Exception {
|
||||
HttpClient client = null;
|
||||
out.printf("%n%s testPostInterrupt(%s, %b)%n", now(), uri, sameClient);
|
||||
for (int i=0; i< ITERATION_COUNT; i++) {
|
||||
if (!sameClient || client == null)
|
||||
client = newHttpClient(sameClient);
|
||||
Thread main = Thread.currentThread();
|
||||
CompletableFuture<Thread> interruptingThread = new CompletableFuture<>();
|
||||
Runnable interrupt = () -> {
|
||||
Thread current = Thread.currentThread();
|
||||
out.printf("%s Interrupting main from: %s (%s)", now(), current, uri);
|
||||
interruptingThread.complete(current);
|
||||
main.interrupt();
|
||||
};
|
||||
Iterable<byte[]> iterable = () -> {
|
||||
var async = random.nextBoolean();
|
||||
if (async) executor.execute(interrupt);
|
||||
else interrupt.run();
|
||||
return List.of(BODY.getBytes(UTF_8)).iterator();
|
||||
};
|
||||
|
||||
HttpRequest req = HttpRequest.newBuilder(URI.create(uri))
|
||||
.POST(HttpRequest.BodyPublishers.ofByteArrays(iterable))
|
||||
.build();
|
||||
String body = null;
|
||||
Exception failed = null;
|
||||
try {
|
||||
body = client.send(req, BodyHandlers.ofString()).body();
|
||||
} catch (Exception x) {
|
||||
failed = x;
|
||||
}
|
||||
|
||||
if (failed instanceof InterruptedException) {
|
||||
out.println("Got expected exception: " + failed);
|
||||
} else if (failed instanceof IOException) {
|
||||
// that could be OK if the main thread was interrupted
|
||||
// from the main thread: the interrupt status could have
|
||||
// been caught by writing to the socket from the main
|
||||
// thread.
|
||||
if (interruptingThread.get() == main) {
|
||||
out.println("Accepting IOException: " + failed);
|
||||
failed.printStackTrace(out);
|
||||
} else {
|
||||
throw failed;
|
||||
}
|
||||
} else if (failed != null) {
|
||||
assertEquals(body, Stream.of(BODY.split("\\|")).collect(Collectors.joining()));
|
||||
throw failed;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
@BeforeTest
|
||||
public void setup() throws Exception {
|
||||
sslContext = new SimpleSSLContext().get();
|
||||
if (sslContext == null)
|
||||
throw new AssertionError("Unexpected null sslContext");
|
||||
|
||||
// HTTP/1.1
|
||||
HttpTestHandler h1_chunkHandler = new HTTPSlowHandler();
|
||||
InetSocketAddress sa = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0);
|
||||
httpTestServer = HttpTestServer.of(HttpServer.create(sa, 0));
|
||||
httpTestServer.addHandler(h1_chunkHandler, "/http1/x/");
|
||||
httpURI = "http://" + httpTestServer.serverAuthority() + "/http1/x/";
|
||||
|
||||
HttpsServer httpsServer = HttpsServer.create(sa, 0);
|
||||
httpsServer.setHttpsConfigurator(new HttpsConfigurator(sslContext));
|
||||
httpsTestServer = HttpTestServer.of(httpsServer);
|
||||
httpsTestServer.addHandler(h1_chunkHandler, "/https1/x/");
|
||||
httpsURI = "https://" + httpsTestServer.serverAuthority() + "/https1/x/";
|
||||
|
||||
// HTTP/2
|
||||
HttpTestHandler h2_chunkedHandler = new HTTPSlowHandler();
|
||||
|
||||
http2TestServer = HttpTestServer.of(new Http2TestServer("localhost", false, 0));
|
||||
http2TestServer.addHandler(h2_chunkedHandler, "/http2/x/");
|
||||
http2URI = "http://" + http2TestServer.serverAuthority() + "/http2/x/";
|
||||
|
||||
https2TestServer = HttpTestServer.of(new Http2TestServer("localhost", true, sslContext));
|
||||
https2TestServer.addHandler(h2_chunkedHandler, "/https2/x/");
|
||||
https2URI = "https://" + https2TestServer.serverAuthority() + "/https2/x/";
|
||||
|
||||
serverCount.addAndGet(4);
|
||||
httpTestServer.start();
|
||||
httpsTestServer.start();
|
||||
http2TestServer.start();
|
||||
https2TestServer.start();
|
||||
}
|
||||
|
||||
@AfterTest
|
||||
public void teardown() throws Exception {
|
||||
String sharedClientName =
|
||||
sharedClient == null ? null : sharedClient.toString();
|
||||
sharedClient = null;
|
||||
Thread.sleep(100);
|
||||
AssertionError fail = TRACKER.check(500);
|
||||
try {
|
||||
httpTestServer.stop();
|
||||
httpsTestServer.stop();
|
||||
http2TestServer.stop();
|
||||
https2TestServer.stop();
|
||||
} finally {
|
||||
if (fail != null) {
|
||||
if (sharedClientName != null) {
|
||||
System.err.println("Shared client name is: " + sharedClientName);
|
||||
}
|
||||
throw fail;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isThreadInterrupt(HttpTestExchange t) {
|
||||
return t.getRequestURI().getPath().contains("/interruptThread");
|
||||
}
|
||||
|
||||
/**
|
||||
* A handler that slowly sends back a body to give time for the
|
||||
* the request to get cancelled before the body is fully received.
|
||||
*/
|
||||
static class HTTPSlowHandler implements HttpTestHandler {
|
||||
@Override
|
||||
public void handle(HttpTestExchange t) throws IOException {
|
||||
try {
|
||||
out.println("HTTPSlowHandler received request to " + t.getRequestURI());
|
||||
System.err.println("HTTPSlowHandler received request to " + t.getRequestURI());
|
||||
|
||||
boolean isThreadInterrupt = isThreadInterrupt(t);
|
||||
byte[] req;
|
||||
try (InputStream is = t.getRequestBody()) {
|
||||
req = is.readAllBytes();
|
||||
}
|
||||
t.sendResponseHeaders(200, -1); // chunked/variable
|
||||
try (OutputStream os = t.getResponseBody()) {
|
||||
// lets split the response in several chunks...
|
||||
String msg = (req != null && req.length != 0)
|
||||
? new String(req, UTF_8)
|
||||
: BODY;
|
||||
String[] str = msg.split("\\|");
|
||||
for (var s : str) {
|
||||
req = s.getBytes(UTF_8);
|
||||
os.write(req);
|
||||
os.flush();
|
||||
try {
|
||||
Thread.sleep(SERVER_LATENCY);
|
||||
} catch (InterruptedException x) {
|
||||
// OK
|
||||
}
|
||||
out.printf("Server wrote %d bytes%n", req.length);
|
||||
}
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
out.println("HTTPSlowHandler: unexpected exception: " + e);
|
||||
e.printStackTrace();
|
||||
throw e;
|
||||
} finally {
|
||||
out.printf("HTTPSlowHandler reply sent: %s%n", t.getRequestURI());
|
||||
System.err.printf("HTTPSlowHandler reply sent: %s%n", t.getRequestURI());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2017, 2018, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2017, 2020, 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
|
||||
@ -32,6 +32,7 @@ import static java.lang.System.out;
|
||||
|
||||
/**
|
||||
* @test
|
||||
* @bug 8245462
|
||||
* @summary Basic test for interrupted blocking send
|
||||
* @run main/othervm InterruptedBlockingSend
|
||||
*/
|
||||
@ -69,6 +70,15 @@ public class InterruptedBlockingSend {
|
||||
} else {
|
||||
out.println("Caught expected InterruptedException: " + throwable);
|
||||
}
|
||||
|
||||
out.println("Interrupting before send");
|
||||
try {
|
||||
Thread.currentThread().interrupt();
|
||||
client.send(request, BodyHandlers.discarding());
|
||||
throw new AssertionError("Expected InterruptedException not thrown");
|
||||
} catch (InterruptedException x) {
|
||||
out.println("Got expected exception: " + x);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user