1 /*

     2  * Copyright (c) 2015, 2017, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.incubator.http;

    27 

    28 import java.io.EOFException;

    29 import java.lang.System.Logger.Level;

    30 import java.nio.ByteBuffer;

    31 import java.util.concurrent.CompletableFuture;

    32 import java.util.concurrent.CompletionStage;

    33 import java.util.concurrent.Executor;

    34 import java.util.function.BiConsumer;

    35 import java.util.function.Consumer;

    36 import java.util.function.Function;

    37 import jdk.incubator.http.ResponseContent.BodyParser;

    38 import jdk.incubator.http.internal.common.Log;

    39 import static jdk.incubator.http.HttpClient.Version.HTTP_1_1;

    40 import jdk.incubator.http.internal.common.MinimalFuture;

    41 import jdk.incubator.http.internal.common.Utils;

    42 

    43 /**

    44  * Handles a HTTP/1.1 response (headers + body).

    45  * There can be more than one of these per Http exchange.

    46  */

    47 class Http1Response<T> {

    48 

    49     private volatile ResponseContent content;

    50     private final HttpRequestImpl request;

    51     private Response response;

    52     private final HttpConnection connection;

    53     private HttpHeaders headers;

    54     private int responseCode;

    55     private final Http1Exchange<T> exchange;

    56     private boolean return2Cache; // return connection to cache when finished

    57     private final HeadersReader headersReader; // used to read the headers

    58     private final BodyReader bodyReader; // used to read the body

    59     private final Http1AsyncReceiver asyncReceiver;

    60     private volatile EOFException eof;

    61     // max number of bytes of (fixed length) body to ignore on redirect

    62     private final static int MAX_IGNORE = 1024;

    63 

    64     // Revisit: can we get rid of this?

    65     static enum State {INITIAL, READING_HEADERS, READING_BODY, DONE}

    66     private volatile State readProgress = State.INITIAL;

    67     static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.

    68     final System.Logger  debug = Utils.getDebugLogger(this.getClass()::getSimpleName, DEBUG);

    69 

    70 

    71     Http1Response(HttpConnection conn,

    72                   Http1Exchange<T> exchange,

    73                   Http1AsyncReceiver asyncReceiver) {

    74         this.readProgress = State.INITIAL;

    75         this.request = exchange.request();

    76         this.exchange = exchange;

    77         this.connection = conn;

    78         this.asyncReceiver = asyncReceiver;

    79         headersReader = new HeadersReader(this::advance);

    80         bodyReader = new BodyReader(this::advance);

    81     }

    82 

    83    public CompletableFuture<Response> readHeadersAsync(Executor executor) {

    84         debug.log(Level.DEBUG, () -> "Reading Headers: (remaining: "

    85                 + asyncReceiver.remaining() +") "  + readProgress);

    86         // with expect continue we will resume reading headers + body.

    87         asyncReceiver.unsubscribe(bodyReader);

    88         bodyReader.reset();

    89         Http1HeaderParser hd = new Http1HeaderParser();

    90         readProgress = State.READING_HEADERS;

    91         headersReader.start(hd);

    92         asyncReceiver.subscribe(headersReader);

    93         CompletableFuture<State> cf = headersReader.completion();

    94         assert cf != null : "parsing not started";

    95 

    96         Function<State, Response> lambda = (State completed) -> {

    97                 assert completed == State.READING_HEADERS;

    98                 debug.log(Level.DEBUG, () ->

    99                             "Reading Headers: creating Response object;"

   100                             + " state is now " + readProgress);

   101                 asyncReceiver.unsubscribe(headersReader);

   102                 responseCode = hd.responseCode();

   103                 headers = hd.headers();

   104 

   105                 response = new Response(request,

   106                                         exchange.getExchange(),

   107                                         headers,

   108                                         responseCode,

   109                                         HTTP_1_1);

   110                 return response;

   111             };

   112 

   113         if (executor != null) {

   114             return cf.thenApplyAsync(lambda, executor);

   115         } else {

   116             return cf.thenApply(lambda);

   117         }

   118     }

   119 

   120     private boolean finished;

   121 

   122     synchronized void completed() {

   123         finished = true;

   124     }

   125 

   126     synchronized boolean finished() {

   127         return finished;

   128     }

   129 

   130     int fixupContentLen(int clen) {

   131         if (request.method().equalsIgnoreCase("HEAD")) {

   132             return 0;

   133         }

   134         if (clen == -1) {

   135             if (headers.firstValue("Transfer-encoding").orElse("")

   136                        .equalsIgnoreCase("chunked")) {

   137                 return -1;

   138             }

   139             return 0;

   140         }

   141         return clen;

   142     }

   143 

   144     /**

   145      * Read up to MAX_IGNORE bytes discarding

   146      */

   147     public CompletableFuture<Void> ignoreBody(Executor executor) {

   148         int clen = (int)headers.firstValueAsLong("Content-Length").orElse(-1);

   149         if (clen == -1 || clen > MAX_IGNORE) {

   150             connection.close();

   151             return MinimalFuture.completedFuture(null); // not treating as error

   152         } else {

   153             return readBody(HttpResponse.BodySubscriber.discard((Void)null), true, executor);

   154         }

   155     }

   156 

   157     public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,

   158                                          boolean return2Cache,

   159                                          Executor executor) {

   160         this.return2Cache = return2Cache;

   161         final HttpResponse.BodySubscriber<U> pusher = p;

   162         final CompletionStage<U> bodyCF = p.getBody();

   163         final CompletableFuture<U> cf = MinimalFuture.of(bodyCF);

   164 

   165         int clen0 = (int)headers.firstValueAsLong("Content-Length").orElse(-1);

   166 

   167         final int clen = fixupContentLen(clen0);

   168 

   169         // expect-continue reads headers and body twice.

   170         // if we reach here, we must reset the headersReader state.

   171         asyncReceiver.unsubscribe(headersReader);

   172         headersReader.reset();

   173 

   174         executor.execute(() -> {

   175             try {

   176                 HttpClientImpl client = connection.client();

   177                 content = new ResponseContent(

   178                         connection, clen, headers, pusher,

   179                         this::onFinished

   180                 );

   181                 if (cf.isCompletedExceptionally()) {

   182                     // if an error occurs during subscription

   183                     connection.close();

   184                     return;

   185                 }

   186                 // increment the reference count on the HttpClientImpl

   187                 // to prevent the SelectorManager thread from exiting until

   188                 // the body is fully read.

   189                 client.reference();

   190                 bodyReader.start(content.getBodyParser(

   191                     (t) -> {

   192                         try {

   193                             if (t != null) {

   194                                 pusher.onError(t);

   195                                 connection.close();

   196                                 if (!cf.isDone())

   197                                     cf.completeExceptionally(t);

   198                             }

   199                         } finally {

   200                             // decrement the reference count on the HttpClientImpl

   201                             // to allow the SelectorManager thread to exit if no

   202                             // other operation is pending and the facade is no

   203                             // longer referenced.

   204                             client.unreference();

   205                             bodyReader.onComplete(t);

   206                         }

   207                     }));

   208                 CompletableFuture<State> bodyReaderCF = bodyReader.completion();

   209                 asyncReceiver.subscribe(bodyReader);

   210                 assert bodyReaderCF != null : "parsing not started";

   211                 // Make sure to keep a reference to asyncReceiver from

   212                 // within this

   213                 CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s,t) ->  {

   214                     t = Utils.getCompletionCause(t);

   215                     try {

   216                         if (t != null) {

   217                             debug.log(Level.DEBUG, () ->

   218                                     "Finished reading body: " + s);

   219                             assert s == State.READING_BODY;

   220                         }

   221                         if (t != null && !cf.isDone()) {

   222                             pusher.onError(t);

   223                             cf.completeExceptionally(t);

   224                         }

   225                     } catch (Throwable x) {

   226                         // not supposed to happen

   227                         asyncReceiver.onReadError(x);

   228                     }

   229                 });

   230                 connection.addTrailingOperation(trailingOp);

   231             } catch (Throwable t) {

   232                debug.log(Level.DEBUG, () -> "Failed reading body: " + t);

   233                 try {

   234                     if (!cf.isDone()) {

   235                         pusher.onError(t);

   236                         cf.completeExceptionally(t);

   237                     }

   238                 } finally {

   239                     asyncReceiver.onReadError(t);

   240                 }

   241             }

   242         });

   243         return cf;

   244     }

   245 

   246 

   247     private void onFinished() {

   248         asyncReceiver.clear();

   249         if (return2Cache) {

   250             Log.logTrace("Attempting to return connection to the pool: {0}", connection);

   251             // TODO: need to do something here?

   252             // connection.setAsyncCallbacks(null, null, null);

   253 

   254             // don't return the connection to the cache if EOF happened.

   255             debug.log(Level.DEBUG, () -> connection.getConnectionFlow()

   256                                    + ": return to HTTP/1.1 pool");

   257             connection.closeOrReturnToCache(eof == null ? headers : null);

   258         }

   259     }

   260 

   261     HttpHeaders responseHeaders() {

   262         return headers;

   263     }

   264 

   265     int responseCode() {

   266         return responseCode;

   267     }

   268 

   269 // ================ Support for plugging into Http1Receiver   =================

   270 // ============================================================================

   271 

   272     // Callback: Error receiver: Consumer of Throwable.

   273     void onReadError(Throwable t) {

   274         Log.logError(t);

   275         Receiver<?> receiver = receiver(readProgress);

   276         if (t instanceof EOFException) {

   277             debug.log(Level.DEBUG, "onReadError: received EOF");

   278             eof = (EOFException) t;

   279         }

   280         CompletableFuture<?> cf = receiver == null ? null : receiver.completion();

   281         debug.log(Level.DEBUG, () -> "onReadError: cf is "

   282                 + (cf == null  ? "null"

   283                 : (cf.isDone() ? "already completed"

   284                                : "not yet completed")));

   285         if (cf != null && !cf.isDone()) cf.completeExceptionally(t);

   286         else { debug.log(Level.DEBUG, "onReadError", t); }

   287         debug.log(Level.DEBUG, () -> "closing connection: cause is " + t);

   288         connection.close();

   289     }

   290 

   291     // ========================================================================

   292 

   293     private State advance(State previous) {

   294         assert readProgress == previous;

   295         switch(previous) {

   296             case READING_HEADERS:

   297                 asyncReceiver.unsubscribe(headersReader);

   298                 return readProgress = State.READING_BODY;

   299             case READING_BODY:

   300                 asyncReceiver.unsubscribe(bodyReader);

   301                 return readProgress = State.DONE;

   302             default:

   303                 throw new InternalError("can't advance from " + previous);

   304         }

   305     }

   306 

   307     Receiver<?> receiver(State state) {

   308         switch(state) {

   309             case READING_HEADERS: return headersReader;

   310             case READING_BODY: return bodyReader;

   311             default: return null;

   312         }

   313 

   314     }

   315 

   316     static abstract class Receiver<T>

   317             implements Http1AsyncReceiver.Http1AsyncDelegate {

   318         abstract void start(T parser);

   319         abstract CompletableFuture<State> completion();

   320         // accepts a buffer from upstream.

   321         // this should be implemented as a simple call to

   322         // accept(ref, parser, cf)

   323         public abstract boolean tryAsyncReceive(ByteBuffer buffer);

   324         public abstract void onReadError(Throwable t);

   325         // handle a byte buffer received from upstream.

   326         // this method should set the value of Http1Response.buffer

   327         // to ref.get() before beginning parsing.

   328         abstract void handle(ByteBuffer buf, T parser,

   329                              CompletableFuture<State> cf);

   330         // resets this objects state so that it can be reused later on

   331         // typically puts the reference to parser and completion to null

   332         abstract void reset();

   333 

   334         // accepts a byte buffer received from upstream

   335         // returns true if the buffer is fully parsed and more data can

   336         // be accepted, false otherwise.

   337         final boolean accept(ByteBuffer buf, T parser,

   338                 CompletableFuture<State> cf) {

   339             if (cf == null || parser == null || cf.isDone()) return false;

   340             handle(buf, parser, cf);

   341             return !cf.isDone();

   342         }

   343         public abstract void onSubscribe(AbstractSubscription s);

   344         public abstract AbstractSubscription subscription();

   345 

   346     }

   347 

   348     // Invoked with each new ByteBuffer when reading headers...

   349     final class HeadersReader extends Receiver<Http1HeaderParser> {

   350         final Consumer<State> onComplete;

   351         volatile Http1HeaderParser parser;

   352         volatile CompletableFuture<State> cf;

   353         volatile long count; // bytes parsed (for debug)

   354         volatile AbstractSubscription subscription;

   355 

   356         HeadersReader(Consumer<State> onComplete) {

   357             this.onComplete = onComplete;

   358         }

   359 

   360         @Override

   361         public AbstractSubscription subscription() {

   362             return subscription;

   363         }

   364 

   365         @Override

   366         public void onSubscribe(AbstractSubscription s) {

   367             this.subscription = s;

   368             s.request(1);

   369         }

   370 

   371         @Override

   372         void reset() {

   373             cf = null;

   374             parser = null;

   375             count = 0;

   376             subscription = null;

   377         }

   378 

   379         // Revisit: do we need to support restarting?

   380         @Override

   381         final void start(Http1HeaderParser hp) {

   382             count = 0;

   383             cf = new MinimalFuture<>();

   384             parser = hp;

   385         }

   386 

   387         @Override

   388         CompletableFuture<State> completion() {

   389             return cf;

   390         }

   391 

   392         @Override

   393         public final boolean tryAsyncReceive(ByteBuffer ref) {

   394             boolean hasDemand = subscription.demand().tryDecrement();

   395             assert hasDemand;

   396             boolean needsMore = accept(ref, parser, cf);

   397             if (needsMore) subscription.request(1);

   398             return needsMore;

   399         }

   400 

   401         @Override

   402         public final void onReadError(Throwable t) {

   403             Http1Response.this.onReadError(t);

   404         }

   405 

   406         @Override

   407         final void handle(ByteBuffer b,

   408                           Http1HeaderParser parser,

   409                           CompletableFuture<State> cf) {

   410             assert cf != null : "parsing not started";

   411             assert parser != null : "no parser";

   412             try {

   413                 count += b.remaining();

   414                 debug.log(Level.DEBUG, () -> "Sending " + b.remaining()

   415                         + "/" + b.capacity() + " bytes to header parser");

   416                 if (parser.parse(b)) {

   417                     count -= b.remaining();

   418                     debug.log(Level.DEBUG, () ->

   419                             "Parsing headers completed. bytes=" + count);

   420                     onComplete.accept(State.READING_HEADERS);

   421                     cf.complete(State.READING_HEADERS);

   422                 }

   423             } catch (Throwable t) {

   424                 debug.log(Level.DEBUG,

   425                         () -> "Header parser failed to handle buffer: " + t);

   426                 cf.completeExceptionally(t);

   427             }

   428         }

   429     }

   430 

   431     // Invoked with each new ByteBuffer when reading bodies...

   432     final class BodyReader extends Receiver<BodyParser> {

   433         final Consumer<State> onComplete;

   434         volatile BodyParser parser;

   435         volatile CompletableFuture<State> cf;

   436         volatile AbstractSubscription subscription;

   437         BodyReader(Consumer<State> onComplete) {

   438             this.onComplete = onComplete;

   439         }

   440 

   441         @Override

   442         void reset() {

   443             parser = null;

   444             cf = null;

   445             subscription = null;

   446         }

   447 

   448         // Revisit: do we need to support restarting?

   449         @Override

   450         final void start(BodyParser parser) {

   451             cf = new MinimalFuture<>();

   452             this.parser = parser;

   453         }

   454 

   455         @Override

   456         CompletableFuture<State> completion() {

   457             return cf;

   458         }

   459 

   460         @Override

   461         public final boolean tryAsyncReceive(ByteBuffer b) {

   462             return accept(b, parser, cf);

   463         }

   464 

   465         @Override

   466         public final void onReadError(Throwable t) {

   467             Http1Response.this.onReadError(t);

   468         }

   469 

   470         @Override

   471         public AbstractSubscription subscription() {

   472             return subscription;

   473         }

   474 

   475         @Override

   476         public void onSubscribe(AbstractSubscription s) {

   477             this.subscription = s;

   478             parser.onSubscribe(s);

   479         }

   480 

   481         @Override

   482         final void handle(ByteBuffer b,

   483                           BodyParser parser,

   484                           CompletableFuture<State> cf) {

   485             assert cf != null : "parsing not started";

   486             assert parser != null : "no parser";

   487             try {

   488                 debug.log(Level.DEBUG, () -> "Sending " + b.remaining()

   489                         + "/" + b.capacity() + " bytes to body parser");

   490                 parser.accept(b);

   491             } catch (Throwable t) {

   492                 debug.log(Level.DEBUG,

   493                         () -> "Body parser failed to handle buffer: " + t);

   494                 if (!cf.isDone()) {

   495                     cf.completeExceptionally(t);

   496                 }

   497             }

   498         }

   499 

   500         final void onComplete(Throwable closedExceptionally) {

   501             if (cf.isDone()) return;

   502             if (closedExceptionally != null) {

   503                 cf.completeExceptionally(closedExceptionally);

   504             } else {

   505                 onComplete.accept(State.READING_BODY);

   506                 cf.complete(State.READING_BODY);

   507             }

   508         }

   509 

   510         @Override

   511         public String toString() {

   512             return super.toString() + "/parser=" + String.valueOf(parser);

   513         }

   514 

   515     }

   516 }