--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/internal/Http1Response.java Tue Feb 06 14:10:28 2018 +0000
@@ -0,0 +1,518 @@
+/*
+ * Copyright (c) 2015, 2018, 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.incubator.http.internal;
+
+import java.io.EOFException;
+import java.lang.System.Logger.Level;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import jdk.incubator.http.HttpHeaders;
+import jdk.incubator.http.HttpResponse;
+import jdk.incubator.http.internal.ResponseContent.BodyParser;
+import jdk.incubator.http.internal.common.Log;
+import jdk.incubator.http.internal.common.MinimalFuture;
+import jdk.incubator.http.internal.common.Utils;
+import static jdk.incubator.http.HttpClient.Version.HTTP_1_1;
+
+/**
+ * Handles a HTTP/1.1 response (headers + body).
+ * There can be more than one of these per Http exchange.
+ */
+class Http1Response<T> {
+
+ private volatile ResponseContent content;
+ private final HttpRequestImpl request;
+ private Response response;
+ private final HttpConnection connection;
+ private HttpHeaders headers;
+ private int responseCode;
+ private final Http1Exchange<T> exchange;
+ private boolean return2Cache; // return connection to cache when finished
+ private final HeadersReader headersReader; // used to read the headers
+ private final BodyReader bodyReader; // used to read the body
+ private final Http1AsyncReceiver asyncReceiver;
+ private volatile EOFException eof;
+ // max number of bytes of (fixed length) body to ignore on redirect
+ private final static int MAX_IGNORE = 1024;
+
+ // Revisit: can we get rid of this?
+ static enum State {INITIAL, READING_HEADERS, READING_BODY, DONE}
+ private volatile State readProgress = State.INITIAL;
+ static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.
+ final System.Logger debug = Utils.getDebugLogger(this.getClass()::getSimpleName, DEBUG);
+
+
+ Http1Response(HttpConnection conn,
+ Http1Exchange<T> exchange,
+ Http1AsyncReceiver asyncReceiver) {
+ this.readProgress = State.INITIAL;
+ this.request = exchange.request();
+ this.exchange = exchange;
+ this.connection = conn;
+ this.asyncReceiver = asyncReceiver;
+ headersReader = new HeadersReader(this::advance);
+ bodyReader = new BodyReader(this::advance);
+ }
+
+ public CompletableFuture<Response> readHeadersAsync(Executor executor) {
+ debug.log(Level.DEBUG, () -> "Reading Headers: (remaining: "
+ + asyncReceiver.remaining() +") " + readProgress);
+ // with expect continue we will resume reading headers + body.
+ asyncReceiver.unsubscribe(bodyReader);
+ bodyReader.reset();
+ Http1HeaderParser hd = new Http1HeaderParser();
+ readProgress = State.READING_HEADERS;
+ headersReader.start(hd);
+ asyncReceiver.subscribe(headersReader);
+ CompletableFuture<State> cf = headersReader.completion();
+ assert cf != null : "parsing not started";
+
+ Function<State, Response> lambda = (State completed) -> {
+ assert completed == State.READING_HEADERS;
+ debug.log(Level.DEBUG, () ->
+ "Reading Headers: creating Response object;"
+ + " state is now " + readProgress);
+ asyncReceiver.unsubscribe(headersReader);
+ responseCode = hd.responseCode();
+ headers = hd.headers();
+
+ response = new Response(request,
+ exchange.getExchange(),
+ headers,
+ responseCode,
+ HTTP_1_1);
+ return response;
+ };
+
+ if (executor != null) {
+ return cf.thenApplyAsync(lambda, executor);
+ } else {
+ return cf.thenApply(lambda);
+ }
+ }
+
+ private boolean finished;
+
+ synchronized void completed() {
+ finished = true;
+ }
+
+ synchronized boolean finished() {
+ return finished;
+ }
+
+ int fixupContentLen(int clen) {
+ if (request.method().equalsIgnoreCase("HEAD")) {
+ return 0;
+ }
+ if (clen == -1) {
+ if (headers.firstValue("Transfer-encoding").orElse("")
+ .equalsIgnoreCase("chunked")) {
+ return -1;
+ }
+ return 0;
+ }
+ return clen;
+ }
+
+ /**
+ * Read up to MAX_IGNORE bytes discarding
+ */
+ public CompletableFuture<Void> ignoreBody(Executor executor) {
+ int clen = (int)headers.firstValueAsLong("Content-Length").orElse(-1);
+ if (clen == -1 || clen > MAX_IGNORE) {
+ connection.close();
+ return MinimalFuture.completedFuture(null); // not treating as error
+ } else {
+ return readBody(HttpResponse.BodySubscriber.discard((Void)null), true, executor);
+ }
+ }
+
+ public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,
+ boolean return2Cache,
+ Executor executor) {
+ this.return2Cache = return2Cache;
+ final HttpResponse.BodySubscriber<U> pusher = p;
+ final CompletionStage<U> bodyCF = p.getBody();
+ final CompletableFuture<U> cf = MinimalFuture.of(bodyCF);
+
+ int clen0 = (int)headers.firstValueAsLong("Content-Length").orElse(-1);
+
+ final int clen = fixupContentLen(clen0);
+
+ // expect-continue reads headers and body twice.
+ // if we reach here, we must reset the headersReader state.
+ asyncReceiver.unsubscribe(headersReader);
+ headersReader.reset();
+
+ executor.execute(() -> {
+ try {
+ HttpClientImpl client = connection.client();
+ content = new ResponseContent(
+ connection, clen, headers, pusher,
+ this::onFinished
+ );
+ if (cf.isCompletedExceptionally()) {
+ // if an error occurs during subscription
+ connection.close();
+ return;
+ }
+ // increment the reference count on the HttpClientImpl
+ // to prevent the SelectorManager thread from exiting until
+ // the body is fully read.
+ client.reference();
+ bodyReader.start(content.getBodyParser(
+ (t) -> {
+ try {
+ if (t != null) {
+ pusher.onError(t);
+ connection.close();
+ if (!cf.isDone())
+ cf.completeExceptionally(t);
+ }
+ } finally {
+ // decrement the reference count on the HttpClientImpl
+ // to allow the SelectorManager thread to exit if no
+ // other operation is pending and the facade is no
+ // longer referenced.
+ client.unreference();
+ bodyReader.onComplete(t);
+ }
+ }));
+ CompletableFuture<State> bodyReaderCF = bodyReader.completion();
+ asyncReceiver.subscribe(bodyReader);
+ assert bodyReaderCF != null : "parsing not started";
+ // Make sure to keep a reference to asyncReceiver from
+ // within this
+ CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s,t) -> {
+ t = Utils.getCompletionCause(t);
+ try {
+ if (t != null) {
+ debug.log(Level.DEBUG, () ->
+ "Finished reading body: " + s);
+ assert s == State.READING_BODY;
+ }
+ if (t != null && !cf.isDone()) {
+ pusher.onError(t);
+ cf.completeExceptionally(t);
+ }
+ } catch (Throwable x) {
+ // not supposed to happen
+ asyncReceiver.onReadError(x);
+ }
+ });
+ connection.addTrailingOperation(trailingOp);
+ } catch (Throwable t) {
+ debug.log(Level.DEBUG, () -> "Failed reading body: " + t);
+ try {
+ if (!cf.isDone()) {
+ pusher.onError(t);
+ cf.completeExceptionally(t);
+ }
+ } finally {
+ asyncReceiver.onReadError(t);
+ }
+ }
+ });
+ return cf;
+ }
+
+
+ private void onFinished() {
+ asyncReceiver.clear();
+ if (return2Cache) {
+ Log.logTrace("Attempting to return connection to the pool: {0}", connection);
+ // TODO: need to do something here?
+ // connection.setAsyncCallbacks(null, null, null);
+
+ // don't return the connection to the cache if EOF happened.
+ debug.log(Level.DEBUG, () -> connection.getConnectionFlow()
+ + ": return to HTTP/1.1 pool");
+ connection.closeOrReturnToCache(eof == null ? headers : null);
+ }
+ }
+
+ HttpHeaders responseHeaders() {
+ return headers;
+ }
+
+ int responseCode() {
+ return responseCode;
+ }
+
+// ================ Support for plugging into Http1Receiver =================
+// ============================================================================
+
+ // Callback: Error receiver: Consumer of Throwable.
+ void onReadError(Throwable t) {
+ Log.logError(t);
+ Receiver<?> receiver = receiver(readProgress);
+ if (t instanceof EOFException) {
+ debug.log(Level.DEBUG, "onReadError: received EOF");
+ eof = (EOFException) t;
+ }
+ CompletableFuture<?> cf = receiver == null ? null : receiver.completion();
+ debug.log(Level.DEBUG, () -> "onReadError: cf is "
+ + (cf == null ? "null"
+ : (cf.isDone() ? "already completed"
+ : "not yet completed")));
+ if (cf != null && !cf.isDone()) cf.completeExceptionally(t);
+ else { debug.log(Level.DEBUG, "onReadError", t); }
+ debug.log(Level.DEBUG, () -> "closing connection: cause is " + t);
+ connection.close();
+ }
+
+ // ========================================================================
+
+ private State advance(State previous) {
+ assert readProgress == previous;
+ switch(previous) {
+ case READING_HEADERS:
+ asyncReceiver.unsubscribe(headersReader);
+ return readProgress = State.READING_BODY;
+ case READING_BODY:
+ asyncReceiver.unsubscribe(bodyReader);
+ return readProgress = State.DONE;
+ default:
+ throw new InternalError("can't advance from " + previous);
+ }
+ }
+
+ Receiver<?> receiver(State state) {
+ switch(state) {
+ case READING_HEADERS: return headersReader;
+ case READING_BODY: return bodyReader;
+ default: return null;
+ }
+
+ }
+
+ static abstract class Receiver<T>
+ implements Http1AsyncReceiver.Http1AsyncDelegate {
+ abstract void start(T parser);
+ abstract CompletableFuture<State> completion();
+ // accepts a buffer from upstream.
+ // this should be implemented as a simple call to
+ // accept(ref, parser, cf)
+ public abstract boolean tryAsyncReceive(ByteBuffer buffer);
+ public abstract void onReadError(Throwable t);
+ // handle a byte buffer received from upstream.
+ // this method should set the value of Http1Response.buffer
+ // to ref.get() before beginning parsing.
+ abstract void handle(ByteBuffer buf, T parser,
+ CompletableFuture<State> cf);
+ // resets this objects state so that it can be reused later on
+ // typically puts the reference to parser and completion to null
+ abstract void reset();
+
+ // accepts a byte buffer received from upstream
+ // returns true if the buffer is fully parsed and more data can
+ // be accepted, false otherwise.
+ final boolean accept(ByteBuffer buf, T parser,
+ CompletableFuture<State> cf) {
+ if (cf == null || parser == null || cf.isDone()) return false;
+ handle(buf, parser, cf);
+ return !cf.isDone();
+ }
+ public abstract void onSubscribe(AbstractSubscription s);
+ public abstract AbstractSubscription subscription();
+
+ }
+
+ // Invoked with each new ByteBuffer when reading headers...
+ final class HeadersReader extends Receiver<Http1HeaderParser> {
+ final Consumer<State> onComplete;
+ volatile Http1HeaderParser parser;
+ volatile CompletableFuture<State> cf;
+ volatile long count; // bytes parsed (for debug)
+ volatile AbstractSubscription subscription;
+
+ HeadersReader(Consumer<State> onComplete) {
+ this.onComplete = onComplete;
+ }
+
+ @Override
+ public AbstractSubscription subscription() {
+ return subscription;
+ }
+
+ @Override
+ public void onSubscribe(AbstractSubscription s) {
+ this.subscription = s;
+ s.request(1);
+ }
+
+ @Override
+ void reset() {
+ cf = null;
+ parser = null;
+ count = 0;
+ subscription = null;
+ }
+
+ // Revisit: do we need to support restarting?
+ @Override
+ final void start(Http1HeaderParser hp) {
+ count = 0;
+ cf = new MinimalFuture<>();
+ parser = hp;
+ }
+
+ @Override
+ CompletableFuture<State> completion() {
+ return cf;
+ }
+
+ @Override
+ public final boolean tryAsyncReceive(ByteBuffer ref) {
+ boolean hasDemand = subscription.demand().tryDecrement();
+ assert hasDemand;
+ boolean needsMore = accept(ref, parser, cf);
+ if (needsMore) subscription.request(1);
+ return needsMore;
+ }
+
+ @Override
+ public final void onReadError(Throwable t) {
+ Http1Response.this.onReadError(t);
+ }
+
+ @Override
+ final void handle(ByteBuffer b,
+ Http1HeaderParser parser,
+ CompletableFuture<State> cf) {
+ assert cf != null : "parsing not started";
+ assert parser != null : "no parser";
+ try {
+ count += b.remaining();
+ debug.log(Level.DEBUG, () -> "Sending " + b.remaining()
+ + "/" + b.capacity() + " bytes to header parser");
+ if (parser.parse(b)) {
+ count -= b.remaining();
+ debug.log(Level.DEBUG, () ->
+ "Parsing headers completed. bytes=" + count);
+ onComplete.accept(State.READING_HEADERS);
+ cf.complete(State.READING_HEADERS);
+ }
+ } catch (Throwable t) {
+ debug.log(Level.DEBUG,
+ () -> "Header parser failed to handle buffer: " + t);
+ cf.completeExceptionally(t);
+ }
+ }
+ }
+
+ // Invoked with each new ByteBuffer when reading bodies...
+ final class BodyReader extends Receiver<BodyParser> {
+ final Consumer<State> onComplete;
+ volatile BodyParser parser;
+ volatile CompletableFuture<State> cf;
+ volatile AbstractSubscription subscription;
+ BodyReader(Consumer<State> onComplete) {
+ this.onComplete = onComplete;
+ }
+
+ @Override
+ void reset() {
+ parser = null;
+ cf = null;
+ subscription = null;
+ }
+
+ // Revisit: do we need to support restarting?
+ @Override
+ final void start(BodyParser parser) {
+ cf = new MinimalFuture<>();
+ this.parser = parser;
+ }
+
+ @Override
+ CompletableFuture<State> completion() {
+ return cf;
+ }
+
+ @Override
+ public final boolean tryAsyncReceive(ByteBuffer b) {
+ return accept(b, parser, cf);
+ }
+
+ @Override
+ public final void onReadError(Throwable t) {
+ Http1Response.this.onReadError(t);
+ }
+
+ @Override
+ public AbstractSubscription subscription() {
+ return subscription;
+ }
+
+ @Override
+ public void onSubscribe(AbstractSubscription s) {
+ this.subscription = s;
+ parser.onSubscribe(s);
+ }
+
+ @Override
+ final void handle(ByteBuffer b,
+ BodyParser parser,
+ CompletableFuture<State> cf) {
+ assert cf != null : "parsing not started";
+ assert parser != null : "no parser";
+ try {
+ debug.log(Level.DEBUG, () -> "Sending " + b.remaining()
+ + "/" + b.capacity() + " bytes to body parser");
+ parser.accept(b);
+ } catch (Throwable t) {
+ debug.log(Level.DEBUG,
+ () -> "Body parser failed to handle buffer: " + t);
+ if (!cf.isDone()) {
+ cf.completeExceptionally(t);
+ }
+ }
+ }
+
+ final void onComplete(Throwable closedExceptionally) {
+ if (cf.isDone()) return;
+ if (closedExceptionally != null) {
+ cf.completeExceptionally(closedExceptionally);
+ } else {
+ onComplete.accept(State.READING_BODY);
+ cf.complete(State.READING_BODY);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return super.toString() + "/parser=" + String.valueOf(parser);
+ }
+
+ }
+}