8217264: HttpClient: Blocking operations in mapper function do not work as documented
Summary: ensures that a new task is spawned when calling getBody() on a mapping BodySubscriber.
Reviewed-by: chegar
--- a/src/java.net.http/share/classes/jdk/internal/net/http/BufferingSubscriber.java Thu Jan 24 11:29:16 2019 +0100
+++ b/src/java.net.http/share/classes/jdk/internal/net/http/BufferingSubscriber.java Thu Jan 24 12:32:42 2019 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2019, 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
@@ -37,14 +37,14 @@
import java.net.http.HttpResponse.BodySubscriber;
import jdk.internal.net.http.common.Demand;
import jdk.internal.net.http.common.SequentialScheduler;
-import jdk.internal.net.http.common.Utils;
+import jdk.internal.net.http.ResponseSubscribers.TrustedSubscriber;
/**
* A buffering BodySubscriber. When subscribed, accumulates ( buffers ) a given
* amount ( in bytes ) of a publisher's data before pushing it to a downstream
* subscriber.
*/
-public class BufferingSubscriber<T> implements BodySubscriber<T>
+public class BufferingSubscriber<T> implements TrustedSubscriber<T>
{
/** The downstream consumer of the data. */
private final BodySubscriber<T> downstreamSubscriber;
@@ -94,6 +94,11 @@
return buffers.stream().mapToLong(ByteBuffer::remaining).sum();
}
+ @Override
+ public boolean needsExecutor() {
+ return TrustedSubscriber.needsExecutor(downstreamSubscriber);
+ }
+
/**
* Tells whether, or not, there is at least a sufficient number of bytes
* accumulated in the internal buffers. If the subscriber is COMPLETE, and
--- a/src/java.net.http/share/classes/jdk/internal/net/http/Http1AsyncReceiver.java Thu Jan 24 11:29:16 2019 +0100
+++ b/src/java.net.http/share/classes/jdk/internal/net/http/Http1AsyncReceiver.java Thu Jan 24 12:32:42 2019 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2019, 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
@@ -34,6 +34,7 @@
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Flow;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -199,7 +200,11 @@
private void flush() {
ByteBuffer buf;
try {
- assert !client.isSelectorThread() :
+ // we should not be running in the selector here,
+ // except if the custom Executor supplied to the client is
+ // something like (r) -> r.run();
+ assert !client.isSelectorThread()
+ || !(client.theExecutor().delegate() instanceof ExecutorService) :
"Http1AsyncReceiver::flush should not run in the selector: "
+ Thread.currentThread().getName();
--- a/src/java.net.http/share/classes/jdk/internal/net/http/Http1Response.java Thu Jan 24 11:29:16 2019 +0100
+++ b/src/java.net.http/share/classes/jdk/internal/net/http/Http1Response.java Thu Jan 24 12:32:42 2019 +0000
@@ -41,6 +41,7 @@
import java.net.http.HttpResponse;
import jdk.internal.net.http.ResponseContent.BodyParser;
import jdk.internal.net.http.ResponseContent.UnknownLengthBodyParser;
+import jdk.internal.net.http.ResponseSubscribers.TrustedSubscriber;
import jdk.internal.net.http.common.Log;
import jdk.internal.net.http.common.Logger;
import jdk.internal.net.http.common.MinimalFuture;
@@ -293,7 +294,7 @@
* subscribed.
* @param <U> The type of response.
*/
- final static class Http1BodySubscriber<U> implements HttpResponse.BodySubscriber<U> {
+ final static class Http1BodySubscriber<U> implements TrustedSubscriber<U> {
final HttpResponse.BodySubscriber<U> userSubscriber;
final AtomicBoolean completed = new AtomicBoolean();
volatile Throwable withError;
@@ -302,6 +303,11 @@
this.userSubscriber = userSubscriber;
}
+ @Override
+ public boolean needsExecutor() {
+ return TrustedSubscriber.needsExecutor(userSubscriber);
+ }
+
// propagate the error to the user subscriber, even if not
// subscribed yet.
private void propagateError(Throwable t) {
@@ -356,6 +362,7 @@
public CompletionStage<U> getBody() {
return userSubscriber.getBody();
}
+
@Override
public void onSubscribe(Flow.Subscription subscription) {
if (!subscribed) {
@@ -475,18 +482,12 @@
connection.client().unreference();
}
});
- try {
- p.getBody().whenComplete((U u, Throwable t) -> {
- if (t == null)
- cf.complete(u);
- else
- cf.completeExceptionally(t);
- });
- } catch (Throwable t) {
+
+ ResponseSubscribers.getBodyAsync(executor, p, cf, (t) -> {
cf.completeExceptionally(t);
asyncReceiver.setRetryOnError(false);
asyncReceiver.onReadError(t);
- }
+ });
return cf.whenComplete((s,t) -> {
if (t != null) {
--- a/src/java.net.http/share/classes/jdk/internal/net/http/MultiExchange.java Thu Jan 24 11:29:16 2019 +0100
+++ b/src/java.net.http/share/classes/jdk/internal/net/http/MultiExchange.java Thu Jan 24 12:32:42 2019 +0000
@@ -271,9 +271,9 @@
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();
+ CompletionStage<T> cs = ResponseSubscribers.getBodyAsync(executor, bs);
MinimalFuture<HttpResponse<T>> result = new MinimalFuture<>();
cs.whenComplete((nullBody, exception) -> {
if (exception != null)
--- a/src/java.net.http/share/classes/jdk/internal/net/http/ResponseSubscribers.java Thu Jan 24 11:29:16 2019 +0100
+++ b/src/java.net.http/share/classes/jdk/internal/net/http/ResponseSubscribers.java Thu Jan 24 12:32:42 2019 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2019, 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
@@ -30,9 +30,6 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
-import java.lang.System.Logger.Level;
-import java.net.http.HttpHeaders;
-import java.net.http.HttpResponse;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
@@ -50,6 +47,7 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
@@ -67,7 +65,50 @@
public class ResponseSubscribers {
- public static class ConsumerSubscriber implements BodySubscriber<Void> {
+ /**
+ * This interface is used by our BodySubscriber implementations to
+ * declare whether calling getBody() inline is safe, or whether
+ * it needs to be called asynchronously in an executor thread.
+ * Calling getBody() inline is usually safe except when it
+ * might block - which can be the case if the BodySubscriber
+ * is provided by custom code, or if it uses a finisher that
+ * might be called and might block before the last bit is
+ * received (for instance, if a mapping subscriber is used with
+ * a mapper function that maps an InputStream to a GZIPInputStream,
+ * as the the constructor of GZIPInputStream calls read()).
+ * @param <T> The response type.
+ */
+ public interface TrustedSubscriber<T> extends BodySubscriber<T> {
+ /**
+ * Returns true if getBody() should be called asynchronously.
+ * @implSpec The default implementation of this method returns
+ * false.
+ * @return true if getBody() should be called asynchronously.
+ */
+ default boolean needsExecutor() { return false;}
+
+ /**
+ * Returns true if calling {@code bs::getBody} might block
+ * and requires an executor.
+ *
+ * @implNote
+ * In particular this method returns
+ * true if {@code bs} is not a {@code TrustedSubscriber}.
+ * If it is a {@code TrustedSubscriber}, it returns
+ * {@code ((TrustedSubscriber) bs).needsExecutor()}.
+ *
+ * @param bs A BodySubscriber.
+ * @return true if calling {@code bs::getBody} requires using
+ * an executor.
+ */
+ static boolean needsExecutor(BodySubscriber<?> bs) {
+ if (bs instanceof TrustedSubscriber) {
+ return ((TrustedSubscriber) bs).needsExecutor();
+ } else return true;
+ }
+ }
+
+ public static class ConsumerSubscriber implements TrustedSubscriber<Void> {
private final Consumer<Optional<byte[]>> consumer;
private Flow.Subscription subscription;
private final CompletableFuture<Void> result = new MinimalFuture<>();
@@ -122,7 +163,7 @@
* asserts the specific, write, file permissions that were checked during
* the construction of this PathSubscriber.
*/
- public static class PathSubscriber implements BodySubscriber<Path> {
+ public static class PathSubscriber implements TrustedSubscriber<Path> {
private static final FilePermission[] EMPTY_FILE_PERMISSIONS = new FilePermission[0];
@@ -223,7 +264,7 @@
}
}
- public static class ByteArraySubscriber<T> implements BodySubscriber<T> {
+ public static class ByteArraySubscriber<T> implements TrustedSubscriber<T> {
private final Function<byte[], T> finisher;
private final CompletableFuture<T> result = new MinimalFuture<>();
private final List<ByteBuffer> received = new ArrayList<>();
@@ -292,7 +333,7 @@
* An InputStream built on top of the Flow API.
*/
public static class HttpResponseInputStream extends InputStream
- implements BodySubscriber<InputStream>
+ implements TrustedSubscriber<InputStream>
{
final static int MAX_BUFFERS_IN_QUEUE = 1; // lock-step with the producer
@@ -410,6 +451,24 @@
}
@Override
+ public int available() throws IOException {
+ // best effort: returns the number of remaining bytes in
+ // the current buffer if any, or 1 if the current buffer
+ // is null or empty but the queue or current buffer list
+ // are not empty. Returns 0 otherwise.
+ if (closed) return 0;
+ int available = 0;
+ ByteBuffer current = currentBuffer;
+ if (current == LAST_BUFFER) return 0;
+ if (current != null) available = current.remaining();
+ if (available != 0) return available;
+ Iterator<?> iterator = currentListItr;
+ if (iterator != null && iterator.hasNext()) return 1;
+ if (buffers.isEmpty()) return 0;
+ return 1;
+ }
+
+ @Override
public void onSubscribe(Flow.Subscription s) {
try {
if (!subscribed.compareAndSet(false, true)) {
@@ -517,17 +576,19 @@
public static BodySubscriber<Stream<String>> createLineStream(Charset charset) {
Objects.requireNonNull(charset);
BodySubscriber<InputStream> s = new HttpResponseInputStream();
+ // Creates a MappingSubscriber with a trusted finisher that is
+ // trusted not to block.
return new MappingSubscriber<InputStream,Stream<String>>(s,
(InputStream stream) -> {
return new BufferedReader(new InputStreamReader(stream, charset))
.lines().onClose(() -> Utils.close(stream));
- });
+ }, true);
}
/**
* Currently this consumes all of the data and ignores it
*/
- public static class NullSubscriber<T> implements BodySubscriber<T> {
+ public static class NullSubscriber<T> implements TrustedSubscriber<T> {
private final CompletableFuture<T> cf = new MinimalFuture<>();
private final Optional<T> result;
@@ -573,13 +634,16 @@
/** An adapter between {@code BodySubscriber} and {@code Flow.Subscriber}. */
public static final class SubscriberAdapter<S extends Subscriber<? super List<ByteBuffer>>,R>
- implements BodySubscriber<R>
+ implements TrustedSubscriber<R>
{
private final CompletableFuture<R> cf = new MinimalFuture<>();
private final S subscriber;
private final Function<? super S,? extends R> finisher;
private volatile Subscription subscription;
+ // The finisher isn't called until all bytes have been received,
+ // and so shouldn't need an executor. No need to override
+ // TrustedSubscriber::needsExecutor
public SubscriberAdapter(S subscriber, Function<? super S,? extends R> finisher) {
this.subscriber = Objects.requireNonNull(subscriber);
this.finisher = Objects.requireNonNull(finisher);
@@ -647,16 +711,40 @@
* @param <T> the upstream body type
* @param <U> this subscriber's body type
*/
- public static class MappingSubscriber<T,U> implements BodySubscriber<U> {
+ public static class MappingSubscriber<T,U> implements TrustedSubscriber<U> {
private final BodySubscriber<T> upstream;
private final Function<? super T,? extends U> mapper;
+ private final boolean trusted;
public MappingSubscriber(BodySubscriber<T> upstream,
Function<? super T,? extends U> mapper) {
+ this(upstream, mapper, false);
+ }
+
+ // creates a MappingSubscriber with a mapper that is trusted
+ // to not block when called.
+ MappingSubscriber(BodySubscriber<T> upstream,
+ Function<? super T,? extends U> mapper,
+ boolean trusted) {
this.upstream = Objects.requireNonNull(upstream);
this.mapper = Objects.requireNonNull(mapper);
+ this.trusted = trusted;
}
+ // There is no way to know whether a custom mapper function
+ // might block or not - so we should return true unless the
+ // mapper is implemented and trusted by our own code not to
+ // block.
+ @Override
+ public boolean needsExecutor() {
+ return !trusted || TrustedSubscriber.needsExecutor(upstream);
+ }
+
+ // If upstream.getBody() is already completed (case of InputStream),
+ // then calling upstream.getBody().thenApply(mapper) might block
+ // if the mapper blocks. We should probably add a variant of
+ // MappingSubscriber that calls thenApplyAsync instead, but this
+ // needs a new public API point. See needsExecutor() above.
@Override
public CompletionStage<U> getBody() {
return upstream.getBody().thenApply(mapper);
@@ -685,7 +773,7 @@
// A BodySubscriber that returns a Publisher<List<ByteBuffer>>
static class PublishingBodySubscriber
- implements BodySubscriber<Flow.Publisher<List<ByteBuffer>>> {
+ implements TrustedSubscriber<Flow.Publisher<List<ByteBuffer>>> {
private final MinimalFuture<Flow.Subscription>
subscriptionCF = new MinimalFuture<>();
private final MinimalFuture<SubscriberRef>
@@ -894,4 +982,110 @@
return new PublishingBodySubscriber();
}
+
+ /**
+ * Tries to determine whether bs::getBody must be invoked asynchronously,
+ * and if so, uses the provided executor to do it.
+ * If the executor is a {@link HttpClientImpl.DelegatingExecutor},
+ * uses the executor's delegate.
+ * @param e The executor to use if an executor is required.
+ * @param bs The BodySubscriber (trusted or not)
+ * @param <T> The type of the response.
+ * @return A completion stage that completes when the completion
+ * stage returned by bs::getBody completes. This may, or
+ * may not, be the same completion stage.
+ */
+ public static <T> CompletionStage<T> getBodyAsync(Executor e, BodySubscriber<T> bs) {
+ if (TrustedSubscriber.needsExecutor(bs)) {
+ // getBody must be called in the executor
+ return getBodyAsync(e, bs, new MinimalFuture<>());
+ } else {
+ // No executor needed
+ return bs.getBody();
+ }
+ }
+
+ /**
+ * Invokes bs::getBody using the provided executor.
+ * If invoking bs::getBody requires an executor, and the given executor
+ * is a {@link HttpClientImpl.DelegatingExecutor}, then the executor's
+ * delegate is used. If an error occurs anywhere then the given {code cf}
+ * is completed exceptionally (this method does not throw).
+ * @param e The executor that should be used to call bs::getBody
+ * @param bs The BodySubscriber
+ * @param cf A completable future that this function will set up
+ * to complete when the completion stage returned by
+ * bs::getBody completes.
+ * In case of any error while trying to set up the
+ * completion chain, {@code cf} will be completed
+ * exceptionally with that error.
+ * @param <T> The response type.
+ * @return The provided {@code cf}.
+ */
+ public static <T> CompletableFuture<T> getBodyAsync(Executor e,
+ BodySubscriber<T> bs,
+ CompletableFuture<T> cf) {
+ return getBodyAsync(e, bs, cf, cf::completeExceptionally);
+ }
+
+ /**
+ * Invokes bs::getBody using the provided executor.
+ * If invoking bs::getBody requires an executor, and the given executor
+ * is a {@link HttpClientImpl.DelegatingExecutor}, then the executor's
+ * delegate is used.
+ * The provided {@code cf} is completed with the result (exceptional
+ * or not) of the completion stage returned by bs::getBody.
+ * If an error occurs when trying to set up the
+ * completion chain, the provided {@code errorHandler} is invoked,
+ * but {@code cf} is not necessarily affected.
+ * This method does not throw.
+ * @param e The executor that should be used to call bs::getBody
+ * @param bs The BodySubscriber
+ * @param cf A completable future that this function will set up
+ * to complete when the completion stage returned by
+ * bs::getBody completes.
+ * In case of any error while trying to set up the
+ * completion chain, {@code cf} will be completed
+ * exceptionally with that error.
+ * @param errorHandler The handler to invoke if an error is raised
+ * while trying to set up the completion chain.
+ * @param <T> The response type.
+ * @return The provide {@code cf}. If the {@code errorHandler} is
+ * invoked, it is the responsibility of the {@code errorHandler} to
+ * complete the {@code cf}, if needed.
+ */
+ public static <T> CompletableFuture<T> getBodyAsync(Executor e,
+ BodySubscriber<T> bs,
+ CompletableFuture<T> cf,
+ Consumer<Throwable> errorHandler) {
+ assert errorHandler != null;
+ try {
+ assert e != null;
+ assert cf != null;
+
+ if (TrustedSubscriber.needsExecutor(bs)) {
+ e = (e instanceof HttpClientImpl.DelegatingExecutor)
+ ? ((HttpClientImpl.DelegatingExecutor) e).delegate() : e;
+ }
+
+ e.execute(() -> {
+ try {
+ bs.getBody().whenComplete((r, t) -> {
+ if (t != null) {
+ cf.completeExceptionally(t);
+ } else {
+ cf.complete(r);
+ }
+ });
+ } catch (Throwable t) {
+ errorHandler.accept(t);
+ }
+ });
+ return cf;
+
+ } catch (Throwable t) {
+ errorHandler.accept(t);
+ }
+ return cf;
+ }
}
--- a/src/java.net.http/share/classes/jdk/internal/net/http/Stream.java Thu Jan 24 11:29:16 2019 +0100
+++ b/src/java.net.http/share/classes/jdk/internal/net/http/Stream.java Thu Jan 24 12:32:42 2019 +0000
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2019, 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
@@ -333,21 +333,10 @@
// pushes entire response body into response subscriber
// blocking when required by local or remote flow control
CompletableFuture<T> receiveData(BodySubscriber<T> bodySubscriber, Executor executor) {
- responseBodyCF = new MinimalFuture<>();
// We want to allow the subscriber's getBody() method to block so it
// can work with InputStreams. So, we offload execution.
- executor.execute(() -> {
- try {
- bodySubscriber.getBody().whenComplete((T body, Throwable t) -> {
- if (t == null)
- responseBodyCF.complete(body);
- else
- responseBodyCF.completeExceptionally(t);
- });
- } catch(Throwable t) {
- cancelImpl(t);
- }
- });
+ responseBodyCF = ResponseSubscribers.getBodyAsync(executor, bodySubscriber,
+ new MinimalFuture<>(), this::cancelImpl);
if (isCanceled()) {
Throwable t = getCancelCause();
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/test/jdk/java/net/httpclient/GZIPInputStreamTest.java Thu Jan 24 12:32:42 2019 +0000
@@ -0,0 +1,650 @@
+/*
+ * Copyright (c) 2019, 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 8217264
+ * @summary Tests that you can map an InputStream to a GZIPInputStream
+ * @library /test/lib http2/server
+ * @build jdk.test.lib.net.SimpleSSLContext
+ * @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 GZIPInputStreamTest
+ */
+
+import com.sun.net.httpserver.HttpServer;
+import com.sun.net.httpserver.HttpsConfigurator;
+import com.sun.net.httpserver.HttpsServer;
+import jdk.test.lib.net.SimpleSSLContext;
+import org.testng.annotations.AfterTest;
+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.io.UncheckedIOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.net.http.HttpResponse.BodyHandler;
+import java.net.http.HttpResponse.BodyHandlers;
+import java.net.http.HttpResponse.BodySubscribers;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import static java.lang.System.out;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.testng.Assert.assertEquals;
+
+public class GZIPInputStreamTest implements HttpServerAdapters {
+
+ 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 int ITERATION_COUNT = 3;
+ // a shared executor helps reduce the amount of threads created by the test
+ // this test will block if the executor doesn't have at least two threads.
+ static final Executor executor = Executors.newFixedThreadPool(2);
+ static final Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
+
+ public static final String LOREM_IPSUM =
+ "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
+ + "Proin et lorem a sem faucibus finibus. "
+ + "Nam nisl nibh, elementum laoreet rutrum quis, lobortis at sem. "
+ + "Aenean purus libero, vehicula sed auctor ac, finibus commodo massa. "
+ + "Etiam dapibus nisl ex, viverra iaculis sapien suscipit sit amet. "
+ + "Phasellus fringilla id orci sit amet convallis. "
+ + "Nam suscipit tempor felis sed feugiat. "
+ + "Mauris quis viverra justo, vitae vulputate turpis. "
+ + "Ut eu orci eget ante faucibus volutpat quis quis urna. "
+ + "Ut porttitor mattis diam, ac sollicitudin ligula volutpat vel. "
+ + "Quisque pretium leo sed augue lacinia, eu mollis dui tempor.\n\n"
+ + "Nullam at mi porttitor, condimentum enim et, tristique felis. "
+ + "Nulla ante elit, interdum id ante ac, dignissim suscipit urna. "
+ + "Sed rhoncus felis eget placerat tincidunt. "
+ + "Duis pellentesque, eros et laoreet lacinia, urna arcu elementum metus, "
+ + "et tempor nibh ante vel odio. "
+ + "Donec et dolor posuere, sagittis libero sit amet, imperdiet ligula. "
+ + "Sed aliquam nulla congue bibendum hendrerit. "
+ + "Morbi ut tincidunt turpis. "
+ + "Nullam semper ipsum et sem imperdiet, sit amet commodo turpis euismod. "
+ + "Nullam aliquet metus id libero elementum, ut pulvinar urna gravida. "
+ + "Nullam non rhoncus diam. "
+ + "Mauris sagittis bibendum odio, sed accumsan sem ullamcorper ut.\n\n"
+ + "Proin malesuada nisl a quam dignissim rhoncus. "
+ + "Pellentesque vitae dui velit. "
+ + "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
+ + "Vivamus sagittis magna id magna vestibulum, nec lacinia odio maximus. "
+ + "Nunc commodo, nisl non sagittis posuere, tortor ligula accumsan diam, "
+ + "a rhoncus augue velit quis enim. "
+ + "Nulla et dictum mauris. "
+ + "Vivamus et accumsan mauris, et tincidunt nunc.\n\n"
+ + "Nullam non pharetra lectus. "
+ + "Fusce lobortis sapien ante, quis egestas tellus tincidunt efficitur. "
+ + "Proin tempus mollis urna, sit amet congue diam eleifend in. "
+ + "Ut auctor metus ipsum, at porta turpis consectetur sed. "
+ + "Ut malesuada euismod massa, ut elementum nisi mattis eget. "
+ + "Donec ultrices vel dolor at convallis. "
+ + "Nunc eget felis nec nunc faucibus finibus. "
+ + "Curabitur nec auctor metus, sit amet tristique lorem. "
+ + "Donec tempus fringilla suscipit. Cras sit amet ante elit. "
+ + "Ut sodales sagittis eros quis cursus. "
+ + "Maecenas finibus ante quis euismod rutrum. "
+ + "Aenean scelerisque placerat nisi. "
+ + "Fusce porta, nibh vel efficitur sodales, urna eros consequat tellus, "
+ + "at fringilla ex justo in mi. "
+ + "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
+ + "Morbi accumsan, justo nec tincidunt pretium, justo ex consectetur ante, "
+ + "nec euismod diam velit vitae quam.\n\n"
+ + "Vestibulum ante ipsum primis in faucibus orci luctus et "
+ + "ultrices posuere cubilia Curae; "
+ + "Praesent eget consequat nunc, vel dapibus nulla. "
+ + "Maecenas egestas luctus consectetur. "
+ + "Duis lacus risus, sollicitudin sit amet justo sed, "
+ + "ultrices facilisis sapien. "
+ + "Mauris eget fermentum risus. "
+ + "Suspendisse potenti. Nam at tempor risus. "
+ + "Quisque lacus augue, dictum vel interdum quis, interdum et mi. "
+ + "In purus mauris, pellentesque et lectus eget, condimentum pretium odio."
+ + " Donec imperdiet congue laoreet. "
+ + "Cras pharetra hendrerit purus ac efficitur. \n";
+
+
+
+ @DataProvider(name = "variants")
+ public Object[][] variants() {
+ return new Object[][]{
+ { httpURI, false },
+ { httpURI, true },
+ { httpsURI, false },
+ { httpsURI, true },
+ { http2URI, false },
+ { http2URI, true },
+ { https2URI, false },
+ { https2URI, true },
+ };
+ }
+
+ final ReferenceTracker TRACKER = ReferenceTracker.INSTANCE;
+ HttpClient newHttpClient() {
+ return TRACKER.track(HttpClient.newBuilder()
+ .executor(executor)
+ .sslContext(sslContext)
+ .build());
+ }
+
+ HttpClient newSingleThreadClient() {
+ return TRACKER.track(HttpClient.newBuilder()
+ .executor(singleThreadExecutor)
+ .sslContext(sslContext)
+ .build());
+ }
+
+ HttpClient newInLineClient() {
+ return TRACKER.track(HttpClient.newBuilder()
+ .executor((r) -> r.run() )
+ .sslContext(sslContext)
+ .build());
+ }
+
+ @Test(dataProvider = "variants")
+ public void testPlainSyncAsString(String uri, boolean sameClient) throws Exception {
+ out.println("\nSmoke test: verify that the result we get from the server is correct.");
+ out.println("Uses plain send() and `asString` to get the plain string.");
+ out.println("Uses single threaded executor");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newSingleThreadClient(); // should work with 1 single thread
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri +"/txt/LoremIpsum.txt"))
+ .build();
+ BodyHandler<String> handler = BodyHandlers.ofString(UTF_8);
+ HttpResponse<String> response = client.send(req, handler);
+ String lorem = response.body();
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testPlainSyncAsInputStream(String uri, boolean sameClient) throws Exception {
+ out.println("Uses plain send() and `asInputStream` - calls readAllBytes() from main thread");
+ out.println("Uses single threaded executor");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newSingleThreadClient(); // should work with 1 single thread
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/txt/LoremIpsum.txt"))
+ .build();
+ BodyHandler<InputStream> handler = BodyHandlers.ofInputStream();
+ HttpResponse<InputStream> response = client.send(req, handler);
+ String lorem = new String(response.body().readAllBytes(), UTF_8);
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testGZIPSyncAsInputStream(String uri, boolean sameClient) throws Exception {
+ out.println("Uses plain send() and `asInputStream` - " +
+ "creates GZIPInputStream and calls readAllBytes() from main thread");
+ out.println("Uses single threaded executor");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newSingleThreadClient(); // should work with 1 single thread
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/gz/LoremIpsum.txt.gz"))
+ .build();
+ BodyHandler<InputStream> handler = BodyHandlers.ofInputStream();
+ HttpResponse<InputStream> response = client.send(req, handler);
+ GZIPInputStream gz = new GZIPInputStream(response.body());
+ String lorem = new String(gz.readAllBytes(), UTF_8);
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testGZIPSyncAsGZIPInputStream(String uri, boolean sameClient) throws Exception {
+ out.println("Uses plain send() and a mapping subscriber to "+
+ "create the GZIPInputStream. Calls readAllBytes() from main thread");
+ out.println("Uses a fixed thread pool executor with 2 thread");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newHttpClient(); // needs at least 2 threads
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/gz/LoremIpsum.txt.gz"))
+ .build();
+ // This is dangerous, because the finisher will block.
+ // We support this, but the executor must have enough threads.
+ BodyHandler<InputStream> handler = new GZIPBodyHandler();
+ HttpResponse<InputStream> response = client.send(req, handler);
+ String lorem = new String(response.body().readAllBytes(), UTF_8);
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testGZIPSyncAsGZIPInputStreamSupplier(String uri, boolean sameClient) throws Exception {
+ out.println("Uses plain send() and a mapping subscriber to "+
+ "create a Supplier<GZIPInputStream>. Calls Supplier.get() " +
+ "and readAllBytes() from main thread");
+ out.println("Uses a single threaded executor");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newSingleThreadClient(); // should work with 1 single thread
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/gz/LoremIpsum.txt.gz"))
+ .build();
+ // This is dangerous, because the finisher will block.
+ // We support this, but the executor must have enough threads.
+ BodyHandler<Supplier<InputStream>> handler = new BodyHandler<Supplier<InputStream>>() {
+ public HttpResponse.BodySubscriber<Supplier<InputStream>> apply(
+ HttpResponse.ResponseInfo responseInfo)
+ {
+ String contentType = responseInfo.headers().firstValue("Content-Encoding")
+ .orElse("identity");
+ out.println("Content-Encoding: " + contentType);
+ if (contentType.equalsIgnoreCase("gzip")) {
+ // This is dangerous. Blocking in the mapping function can wedge the
+ // response. We do support it provided that there enough thread in
+ // the executor.
+ return BodySubscribers.mapping(BodySubscribers.ofInputStream(),
+ (is) -> (() -> {
+ try {
+ return new GZIPInputStream(is);
+ } catch (IOException io) {
+ throw new UncheckedIOException(io);
+ }
+ }));
+ } else return BodySubscribers.mapping(BodySubscribers.ofInputStream(),
+ (is) -> (() -> is));
+ }
+ };
+ HttpResponse<Supplier<InputStream>> response = client.send(req, handler);
+ String lorem = new String(response.body().get().readAllBytes(), UTF_8);
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testPlainAsyncAsInputStreamBlocks(String uri, boolean sameClient) throws Exception {
+ out.println("Uses sendAsync() and `asInputStream`. Registers a dependent action "+
+ "that calls readAllBytes()");
+ out.println("Uses a fixed thread pool executor with two threads");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newHttpClient(); // needs at least 2 threads
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/txt/LoremIpsum.txt"))
+ .build();
+ BodyHandler<InputStream> handler = BodyHandlers.ofInputStream();
+ CompletableFuture<HttpResponse<InputStream>> responseCF = client.sendAsync(req, handler);
+ // This is dangerous. Blocking in the mapping function can wedge the
+ // response. We do support it provided that there enough threads in
+ // the executor.
+ String lorem = responseCF.thenApply((r) -> {
+ try {
+ return new String(r.body().readAllBytes(), UTF_8);
+ } catch (IOException io) {
+ throw new UncheckedIOException(io);
+ }
+ }).join();
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testGZIPAsyncAsGZIPInputStreamBlocks(String uri, boolean sameClient) throws Exception {
+ out.println("Uses sendAsync() and a mapping subscriber to create a GZIPInputStream. " +
+ "Registers a dependent action that calls readAllBytes()");
+ out.println("Uses a fixed thread pool executor with two threads");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newHttpClient(); // needs at least 2 threads
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/gz/LoremIpsum.txt.gz"))
+ .build();
+ BodyHandler<InputStream> handler = new GZIPBodyHandler();
+ CompletableFuture<HttpResponse<InputStream>> responseCF = client.sendAsync(req, handler);
+ // This is dangerous - we support this, but it can block
+ // if there are not enough threads available.
+ // Correct custom code should use thenApplyAsync instead.
+ String lorem = responseCF.thenApply((r) -> {
+ try {
+ return new String(r.body().readAllBytes(), UTF_8);
+ } catch (IOException io) {
+ throw new UncheckedIOException(io);
+ }
+ }).join();
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testGZIPSyncAsGZIPInputStreamBlocks(String uri, boolean sameClient) throws Exception {
+ out.println("Uses sendAsync() and a mapping subscriber to create a GZIPInputStream," +
+ "which is mapped again using a mapping subscriber " +
+ "to call readAllBytes() and map to String");
+ out.println("Uses a fixed thread pool executor with two threads");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newHttpClient(); // needs at least 2 threads
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/gz/LoremIpsum.txt.gz"))
+ .build();
+ // This is dangerous. Blocking in the mapping function can wedge the
+ // response. We do support it provided that there enough thread in
+ // the executor.
+ BodyHandler<String> handler = new MappingBodyHandler<>(new GZIPBodyHandler(),
+ (InputStream is) -> {
+ try {
+ return new String(is.readAllBytes(), UTF_8);
+ } catch(IOException io) {
+ throw new UncheckedIOException(io);
+ }
+ });
+ HttpResponse<String> response = client.send(req, handler);
+ String lorem = response.body();
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ @Test(dataProvider = "variants")
+ public void testGZIPSyncAsGZIPInputStreamSupplierInline(String uri, boolean sameClient) throws Exception {
+ out.println("Uses plain send() and a mapping subscriber to "+
+ "create a Supplier<GZIPInputStream>. Calls Supplier.get() " +
+ "and readAllBytes() from main thread");
+ out.println("Uses an inline executor (no threads)");
+ HttpClient client = null;
+ for (int i=0; i< ITERATION_COUNT; i++) {
+ if (!sameClient || client == null)
+ client = newInLineClient(); // should even work with no threads
+
+ HttpRequest req = HttpRequest.newBuilder(URI.create(uri + "/gz/LoremIpsum.txt.gz"))
+ .build();
+ // This is dangerous, because the finisher will block.
+ // We support this, but the executor must have enough threads.
+ BodyHandler<Supplier<InputStream>> handler = new BodyHandler<Supplier<InputStream>>() {
+ public HttpResponse.BodySubscriber<Supplier<InputStream>> apply(
+ HttpResponse.ResponseInfo responseInfo)
+ {
+ String contentType = responseInfo.headers().firstValue("Content-Encoding")
+ .orElse("identity");
+ out.println("Content-Encoding: " + contentType);
+ if (contentType.equalsIgnoreCase("gzip")) {
+ // This is dangerous. Blocking in the mapping function can wedge the
+ // response. We do support it provided that there enough thread in
+ // the executor.
+ return BodySubscribers.mapping(BodySubscribers.ofInputStream(),
+ (is) -> (() -> {
+ try {
+ return new GZIPInputStream(is);
+ } catch (IOException io) {
+ throw new UncheckedIOException(io);
+ }
+ }));
+ } else return BodySubscribers.mapping(BodySubscribers.ofInputStream(),
+ (is) -> (() -> is));
+ }
+ };
+ HttpResponse<Supplier<InputStream>> response = client.send(req, handler);
+ String lorem = new String(response.body().get().readAllBytes(), UTF_8);
+ if (!LOREM_IPSUM.equals(lorem)) {
+ out.println("Response doesn't match");
+ out.println("[" + LOREM_IPSUM + "] != [" + lorem + "]");
+ assertEquals(LOREM_IPSUM, lorem);
+ } else {
+ out.println("Received expected response.");
+ }
+ }
+ }
+
+ static final class GZIPBodyHandler implements BodyHandler<InputStream> {
+ @Override
+ public HttpResponse.BodySubscriber<InputStream> apply(HttpResponse.ResponseInfo responseInfo) {
+ String contentType = responseInfo.headers().firstValue("Content-Encoding")
+ .orElse("identity");
+ out.println("Content-Encoding: " + contentType);
+ if (contentType.equalsIgnoreCase("gzip")) {
+ // This is dangerous. Blocking in the mapping function can wedge the
+ // response. We do support it provided that there enough thread in
+ // the executor.
+ return BodySubscribers.mapping(BodySubscribers.ofInputStream(),
+ (is) -> {
+ try {
+ return new GZIPInputStream(is);
+ } catch (IOException io) {
+ throw new UncheckedIOException(io);
+ }
+ });
+ } else return BodySubscribers.ofInputStream();
+ }
+ }
+
+ static final class MappingBodyHandler<T,U> implements BodyHandler<U> {
+ final BodyHandler<T> upstream;
+ final Function<? super T,? extends U> finisher;
+ MappingBodyHandler(BodyHandler<T> upstream, Function<T,U> finisher) {
+ this.upstream = upstream;
+ this.finisher = finisher;
+ }
+ @Override
+ public HttpResponse.BodySubscriber<U> apply(HttpResponse.ResponseInfo responseInfo) {
+ return BodySubscribers.mapping(upstream.apply(responseInfo), finisher);
+ }
+ }
+
+ static String serverAuthority(HttpServer server) {
+ return InetAddress.getLoopbackAddress().getHostName() + ":"
+ + server.getAddress().getPort();
+ }
+
+ @BeforeTest
+ public void setup() throws Exception {
+ sslContext = new SimpleSSLContext().get();
+ if (sslContext == null)
+ throw new AssertionError("Unexpected null sslContext");
+
+ HttpTestHandler plainHandler = new LoremIpsumPlainHandler();
+ HttpTestHandler gzipHandler = new LoremIpsumGZIPHandler();
+
+ // HTTP/1.1
+ InetSocketAddress sa = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0);
+ httpTestServer = HttpTestServer.of(HttpServer.create(sa, 0));
+ httpTestServer.addHandler(plainHandler, "/http1/chunk/txt");
+ httpTestServer.addHandler(gzipHandler, "/http1/chunk/gz");
+ httpURI = "http://" + httpTestServer.serverAuthority() + "/http1/chunk";
+
+ HttpsServer httpsServer = HttpsServer.create(sa, 0);
+ httpsServer.setHttpsConfigurator(new HttpsConfigurator(sslContext));
+ httpsTestServer = HttpTestServer.of(httpsServer);
+ httpsTestServer.addHandler(plainHandler, "/https1/chunk/txt");
+ httpsTestServer.addHandler(gzipHandler, "/https1/chunk/gz");
+ httpsURI = "https://" + httpsTestServer.serverAuthority() + "/https1/chunk";
+
+ // HTTP/2
+ http2TestServer = HttpTestServer.of(new Http2TestServer("localhost", false, 0));
+ http2TestServer.addHandler(plainHandler, "/http2/chunk/txt");
+ http2TestServer.addHandler(gzipHandler, "/http2/chunk/gz");
+ http2URI = "http://" + http2TestServer.serverAuthority() + "/http2/chunk";
+
+ https2TestServer = HttpTestServer.of(new Http2TestServer("localhost", true, sslContext));
+ https2TestServer.addHandler(plainHandler, "/https2/chunk/txt");
+ https2TestServer.addHandler(gzipHandler, "/https2/chunk/gz");
+ https2URI = "https://" + https2TestServer.serverAuthority() + "/https2/chunk";
+
+ httpTestServer.start();
+ httpsTestServer.start();
+ http2TestServer.start();
+ https2TestServer.start();
+ }
+
+ @AfterTest
+ public void teardown() throws Exception {
+ Thread.sleep(100);
+ AssertionError fail = TRACKER.check(500);
+ try {
+ httpTestServer.stop();
+ httpsTestServer.stop();
+ http2TestServer.stop();
+ https2TestServer.stop();
+ } finally {
+ if (fail != null) {
+ throw fail;
+ }
+ }
+ }
+
+
+ static class LoremIpsumPlainHandler implements HttpTestHandler {
+ @Override
+ public void handle(HttpTestExchange t) throws IOException {
+ try {
+ out.println("LoremIpsumPlainHandler received request to " + t.getRequestURI());
+ t.getResponseHeaders().addHeader("Content-Encoding", "identity");
+ t.sendResponseHeaders(200, -1);
+ long size = 0;
+ try (OutputStream os = t.getResponseBody()) {
+ for (String s : LOREM_IPSUM.split("\n")) {
+ byte[] buf = s.getBytes(StandardCharsets.UTF_8);
+ System.out.println("Writing " + (buf.length + 1) + " bytes...");
+ os.write(buf);
+ os.write('\n');
+ os.flush();
+ size += buf.length + 1;
+ }
+ } finally {
+ System.out.println("Sent " + size + " bytes");
+ }
+ } catch (IOException | RuntimeException | Error e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+ }
+
+ static class LoremIpsumGZIPHandler implements HttpTestHandler {
+ @Override
+ public void handle(HttpTestExchange t) throws IOException {
+ try {
+ out.println("LoremIpsumGZIPHandler received request to " + t.getRequestURI());
+ t.getResponseHeaders().addHeader("Content-Encoding", "gzip");
+ t.sendResponseHeaders(200, -1);
+ long size = 0;
+ try (GZIPOutputStream os =
+ new GZIPOutputStream(t.getResponseBody())) {
+ for (String s : LOREM_IPSUM.split("\n")) {
+ byte[] buf = s.getBytes(StandardCharsets.UTF_8);
+ System.out.println("Writing and compressing "
+ + (buf.length + 1) + " uncompressed bytes...");
+ os.write(buf);
+ os.write('\n');
+ os.flush();
+ size += buf.length + 1;
+ }
+ } finally {
+ System.out.println("Sent and compressed " + size + " uncompressed bytes");
+ }
+ } catch (IOException | RuntimeException | Error e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+ }
+
+}