1 /*

     2  * Copyright (c) 2016, 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.IOException;

    29 import java.io.InputStream;

    30 import java.io.UncheckedIOException;

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

    32 import java.net.URI;

    33 import java.nio.ByteBuffer;

    34 import java.nio.channels.FileChannel;

    35 import java.nio.file.Files;

    36 import java.nio.file.OpenOption;

    37 import java.nio.file.Path;

    38 import java.security.AccessControlContext;

    39 import java.security.AccessController;

    40 import java.security.PrivilegedActionException;

    41 import java.security.PrivilegedExceptionAction;

    42 import java.util.ArrayList;

    43 import java.util.Iterator;

    44 import java.util.List;

    45 import java.util.Optional;

    46 import java.util.concurrent.ArrayBlockingQueue;

    47 import java.util.concurrent.BlockingQueue;

    48 import java.util.concurrent.CompletableFuture;

    49 import java.util.concurrent.CompletionStage;

    50 import java.util.concurrent.ConcurrentHashMap;

    51 import java.util.concurrent.Flow;

    52 import java.util.function.Consumer;

    53 import java.util.function.Function;

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

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

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

    57 

    58 class ResponseSubscribers {

    59 

    60     static class ConsumerSubscriber implements HttpResponse.BodySubscriber<Void> {

    61         private final Consumer<Optional<byte[]>> consumer;

    62         private Flow.Subscription subscription;

    63         private final CompletableFuture<Void> result = new MinimalFuture<>();

    64 

    65         ConsumerSubscriber(Consumer<Optional<byte[]>> consumer) {

    66             this.consumer = consumer;

    67         }

    68 

    69         @Override

    70         public CompletionStage<Void> getBody() {

    71             return result;

    72         }

    73 

    74         @Override

    75         public void onSubscribe(Flow.Subscription subscription) {

    76             this.subscription = subscription;

    77             subscription.request(1);

    78         }

    79 

    80         @Override

    81         public void onNext(List<ByteBuffer> items) {

    82             for (ByteBuffer item : items) {

    83                 byte[] buf = new byte[item.remaining()];

    84                 item.get(buf);

    85                 consumer.accept(Optional.of(buf));

    86             }

    87             subscription.request(1);

    88         }

    89 

    90         @Override

    91         public void onError(Throwable throwable) {

    92             result.completeExceptionally(throwable);

    93         }

    94 

    95         @Override

    96         public void onComplete() {

    97             consumer.accept(Optional.empty());

    98             result.complete(null);

    99         }

   100 

   101     }

   102 

   103     static class PathSubscriber implements HttpResponse.BodySubscriber<Path> {

   104 

   105         private final Path file;

   106         private final CompletableFuture<Path> result = new MinimalFuture<>();

   107 

   108         private volatile Flow.Subscription subscription;

   109         private volatile FileChannel out;

   110         private volatile AccessControlContext acc;

   111         private final OpenOption[] options;

   112 

   113         PathSubscriber(Path file, OpenOption... options) {

   114             this.file = file;

   115             this.options = options;

   116         }

   117 

   118         void setAccessControlContext(AccessControlContext acc) {

   119             this.acc = acc;

   120         }

   121 

   122         @Override

   123         public void onSubscribe(Flow.Subscription subscription) {

   124             if (System.getSecurityManager() != null && acc == null)

   125                 throw new InternalError(

   126                         "Unexpected null acc when security manager has been installed");

   127 

   128             this.subscription = subscription;

   129             try {

   130                 PrivilegedExceptionAction<FileChannel> pa =

   131                         () -> FileChannel.open(file, options);

   132                 out = AccessController.doPrivileged(pa, acc);

   133             } catch (PrivilegedActionException pae) {

   134                 Throwable t = pae.getCause() != null ? pae.getCause() : pae;

   135                 result.completeExceptionally(t);

   136                 subscription.cancel();

   137                 return;

   138             }

   139             subscription.request(1);

   140         }

   141 

   142         @Override

   143         public void onNext(List<ByteBuffer> items) {

   144             try {

   145                 out.write(items.toArray(new ByteBuffer[0]));

   146             } catch (IOException ex) {

   147                 Utils.close(out);

   148                 subscription.cancel();

   149                 result.completeExceptionally(ex);

   150             }

   151             subscription.request(1);

   152         }

   153 

   154         @Override

   155         public void onError(Throwable e) {

   156             result.completeExceptionally(e);

   157             Utils.close(out);

   158         }

   159 

   160         @Override

   161         public void onComplete() {

   162             Utils.close(out);

   163             result.complete(file);

   164         }

   165 

   166         @Override

   167         public CompletionStage<Path> getBody() {

   168             return result;

   169         }

   170     }

   171 

   172     static class ByteArraySubscriber<T> implements HttpResponse.BodySubscriber<T> {

   173         private final Function<byte[], T> finisher;

   174         private final CompletableFuture<T> result = new MinimalFuture<>();

   175         private final List<ByteBuffer> received = new ArrayList<>();

   176 

   177         private volatile Flow.Subscription subscription;

   178 

   179         ByteArraySubscriber(Function<byte[],T> finisher) {

   180             this.finisher = finisher;

   181         }

   182 

   183         @Override

   184         public void onSubscribe(Flow.Subscription subscription) {

   185             if (this.subscription != null) {

   186                 subscription.cancel();

   187                 return;

   188             }

   189             this.subscription = subscription;

   190             // We can handle whatever you've got

   191             subscription.request(Long.MAX_VALUE);

   192         }

   193 

   194         @Override

   195         public void onNext(List<ByteBuffer> items) {

   196             // incoming buffers are allocated by http client internally,

   197             // and won't be used anywhere except this place.

   198             // So it's free simply to store them for further processing.

   199             assert Utils.hasRemaining(items);

   200             Utils.accumulateBuffers(received, items);

   201         }

   202 

   203         @Override

   204         public void onError(Throwable throwable) {

   205             received.clear();

   206             result.completeExceptionally(throwable);

   207         }

   208 

   209         static private byte[] join(List<ByteBuffer> bytes) {

   210             int size = Utils.remaining(bytes, Integer.MAX_VALUE);

   211             byte[] res = new byte[size];

   212             int from = 0;

   213             for (ByteBuffer b : bytes) {

   214                 int l = b.remaining();

   215                 b.get(res, from, l);

   216                 from += l;

   217             }

   218             return res;

   219         }

   220 

   221         @Override

   222         public void onComplete() {

   223             try {

   224                 result.complete(finisher.apply(join(received)));

   225                 received.clear();

   226             } catch (IllegalArgumentException e) {

   227                 result.completeExceptionally(e);

   228             }

   229         }

   230 

   231         @Override

   232         public CompletionStage<T> getBody() {

   233             return result;

   234         }

   235     }

   236 

   237     /**

   238      * An InputStream built on top of the Flow API.

   239      */

   240     static class HttpResponseInputStream extends InputStream

   241         implements HttpResponse.BodySubscriber<InputStream>

   242     {

   243         final static boolean DEBUG = Utils.DEBUG;

   244         final static int MAX_BUFFERS_IN_QUEUE = 1;  // lock-step with the producer

   245 

   246         // An immutable ByteBuffer sentinel to mark that the last byte was received.

   247         private static final ByteBuffer LAST_BUFFER = ByteBuffer.wrap(new byte[0]);

   248         private static final List<ByteBuffer> LAST_LIST = List.of(LAST_BUFFER);

   249         private static final System.Logger DEBUG_LOGGER =

   250                 Utils.getDebugLogger("HttpResponseInputStream"::toString, DEBUG);

   251 

   252         // A queue of yet unprocessed ByteBuffers received from the flow API.

   253         private final BlockingQueue<List<ByteBuffer>> buffers;

   254         private volatile Flow.Subscription subscription;

   255         private volatile boolean closed;

   256         private volatile Throwable failed;

   257         private volatile Iterator<ByteBuffer> currentListItr;

   258         private volatile ByteBuffer currentBuffer;

   259 

   260         HttpResponseInputStream() {

   261             this(MAX_BUFFERS_IN_QUEUE);

   262         }

   263 

   264         HttpResponseInputStream(int maxBuffers) {

   265             int capacity = (maxBuffers <= 0 ? MAX_BUFFERS_IN_QUEUE : maxBuffers);

   266             // 1 additional slot needed for LAST_LIST added by onComplete

   267             this.buffers = new ArrayBlockingQueue<>(capacity + 1);

   268         }

   269 

   270         @Override

   271         public CompletionStage<InputStream> getBody() {

   272             // Returns the stream immediately, before the

   273             // response body is received.

   274             // This makes it possible for senAsync().get().body()

   275             // to complete before the response body is received.

   276             return CompletableFuture.completedStage(this);

   277         }

   278 

   279         // Returns the current byte buffer to read from.

   280         // If the current buffer has no remaining data, this method will take the

   281         // next buffer from the buffers queue, possibly blocking until

   282         // a new buffer is made available through the Flow API, or the

   283         // end of the flow has been reached.

   284         private ByteBuffer current() throws IOException {

   285             while (currentBuffer == null || !currentBuffer.hasRemaining()) {

   286                 // Check whether the stream is closed or exhausted

   287                 if (closed || failed != null) {

   288                     throw new IOException("closed", failed);

   289                 }

   290                 if (currentBuffer == LAST_BUFFER) break;

   291 

   292                 try {

   293                     if (currentListItr == null || !currentListItr.hasNext()) {

   294                         // Take a new list of buffers from the queue, blocking

   295                         // if none is available yet...

   296 

   297                         DEBUG_LOGGER.log(Level.DEBUG, "Taking list of Buffers");

   298                         List<ByteBuffer> lb = buffers.take();

   299                         currentListItr = lb.iterator();

   300                         DEBUG_LOGGER.log(Level.DEBUG, "List of Buffers Taken");

   301 

   302                         // Check whether an exception was encountered upstream

   303                         if (closed || failed != null)

   304                             throw new IOException("closed", failed);

   305 

   306                         // Check whether we're done.

   307                         if (lb == LAST_LIST) {

   308                             currentListItr = null;

   309                             currentBuffer = LAST_BUFFER;

   310                             break;

   311                         }

   312 

   313                         // Request another upstream item ( list of buffers )

   314                         Flow.Subscription s = subscription;

   315                         if (s != null) {

   316                             DEBUG_LOGGER.log(Level.DEBUG, "Increased demand by 1");

   317                             s.request(1);

   318                         }

   319                     }

   320                     assert currentListItr != null;

   321                     assert currentListItr.hasNext();

   322                     DEBUG_LOGGER.log(Level.DEBUG, "Next Buffer");

   323                     currentBuffer = currentListItr.next();

   324                 } catch (InterruptedException ex) {

   325                     // continue

   326                 }

   327             }

   328             assert currentBuffer == LAST_BUFFER || currentBuffer.hasRemaining();

   329             return currentBuffer;

   330         }

   331 

   332         @Override

   333         public int read(byte[] bytes, int off, int len) throws IOException {

   334             // get the buffer to read from, possibly blocking if

   335             // none is available

   336             ByteBuffer buffer;

   337             if ((buffer = current()) == LAST_BUFFER) return -1;

   338 

   339             // don't attempt to read more than what is available

   340             // in the current buffer.

   341             int read = Math.min(buffer.remaining(), len);

   342             assert read > 0 && read <= buffer.remaining();

   343 

   344             // buffer.get() will do the boundary check for us.

   345             buffer.get(bytes, off, read);

   346             return read;

   347         }

   348 

   349         @Override

   350         public int read() throws IOException {

   351             ByteBuffer buffer;

   352             if ((buffer = current()) == LAST_BUFFER) return -1;

   353             return buffer.get() & 0xFF;

   354         }

   355 

   356         @Override

   357         public void onSubscribe(Flow.Subscription s) {

   358             if (this.subscription != null) {

   359                 s.cancel();

   360                 return;

   361             }

   362             this.subscription = s;

   363             assert buffers.remainingCapacity() > 1; // should contain at least 2

   364             DEBUG_LOGGER.log(Level.DEBUG, () -> "onSubscribe: requesting "

   365                                 +  Math.max(1, buffers.remainingCapacity() - 1));

   366             s.request(Math.max(1, buffers.remainingCapacity() - 1));

   367         }

   368 

   369         @Override

   370         public void onNext(List<ByteBuffer> t) {

   371             try {

   372                 DEBUG_LOGGER.log(Level.DEBUG, "next item received");

   373                 if (!buffers.offer(t)) {

   374                     throw new IllegalStateException("queue is full");

   375                 }

   376                 DEBUG_LOGGER.log(Level.DEBUG, "item offered");

   377             } catch (Exception ex) {

   378                 failed = ex;

   379                 try {

   380                     close();

   381                 } catch (IOException ex1) {

   382                     // OK

   383                 }

   384             }

   385         }

   386 

   387         @Override

   388         public void onError(Throwable thrwbl) {

   389             subscription = null;

   390             failed = thrwbl;

   391         }

   392 

   393         @Override

   394         public void onComplete() {

   395             subscription = null;

   396             onNext(LAST_LIST);

   397         }

   398 

   399         @Override

   400         public void close() throws IOException {

   401             synchronized (this) {

   402                 if (closed) return;

   403                 closed = true;

   404             }

   405             Flow.Subscription s = subscription;

   406             subscription = null;

   407             if (s != null) {

   408                  s.cancel();

   409             }

   410             super.close();

   411         }

   412 

   413     }

   414 

   415     static class MultiSubscriberImpl<V>

   416         implements HttpResponse.MultiSubscriber<MultiMapResult<V>,V>

   417     {

   418         private final MultiMapResult<V> results;

   419         private final Function<HttpRequest,Optional<HttpResponse.BodyHandler<V>>> pushHandler;

   420         private final Function<HttpRequest,HttpResponse.BodyHandler<V>> requestHandler;

   421         private final boolean completion; // aggregate completes on last PP received or overall completion

   422 

   423         MultiSubscriberImpl(

   424                 Function<HttpRequest,HttpResponse.BodyHandler<V>> requestHandler,

   425                 Function<HttpRequest,Optional<HttpResponse.BodyHandler<V>>> pushHandler, boolean completion) {

   426             this.results = new MultiMapResult<>(new ConcurrentHashMap<>());

   427             this.requestHandler = requestHandler;

   428             this.pushHandler = pushHandler;

   429             this.completion = completion;

   430         }

   431 

   432         @Override

   433         public HttpResponse.BodyHandler<V> onRequest(HttpRequest request) {

   434             CompletableFuture<HttpResponse<V>> cf = MinimalFuture.newMinimalFuture();

   435             results.put(request, cf);

   436             return requestHandler.apply(request);

   437         }

   438 

   439         @Override

   440         public Optional<HttpResponse.BodyHandler<V>> onPushPromise(HttpRequest push) {

   441             CompletableFuture<HttpResponse<V>> cf = MinimalFuture.newMinimalFuture();

   442             results.put(push, cf);

   443             return pushHandler.apply(push);

   444         }

   445 

   446         @Override

   447         public void onResponse(HttpResponse<V> response) {

   448             HttpRequest request = response.request();

   449             CompletableFuture<HttpResponse<V>> cf = results.get(response.request());

   450             cf.complete(response);

   451         }

   452 

   453         @Override

   454         public void onError(HttpRequest request, Throwable t) {

   455             CompletableFuture<HttpResponse<V>> cf = results.get(request);

   456             cf.completeExceptionally(t);

   457         }

   458 

   459         @Override

   460         public CompletableFuture<MultiMapResult<V>> completion(

   461                 CompletableFuture<Void> onComplete, CompletableFuture<Void> onFinalPushPromise) {

   462             if (completion)

   463                 return onComplete.thenApply((ignored)-> results);

   464             else

   465                 return onFinalPushPromise.thenApply((ignored) -> results);

   466         }

   467     }

   468 

   469     static class MultiFile {

   470 

   471         final Path pathRoot;

   472 

   473         MultiFile(Path destination) {

   474             if (!destination.toFile().isDirectory())

   475                 throw new UncheckedIOException(new IOException("destination is not a directory"));

   476             pathRoot = destination;

   477         }

   478 

   479         Optional<HttpResponse.BodyHandler<Path>> handlePush(HttpRequest request) {

   480             final URI uri = request.uri();

   481             String path = uri.getPath();

   482             while (path.startsWith("/"))

   483                 path = path.substring(1);

   484             Path p = pathRoot.resolve(path);

   485             if (Log.trace()) {

   486                 Log.logTrace("Creating file body subscriber for URI={0}, path={1}",

   487                              uri, p);

   488             }

   489             try {

   490                 Files.createDirectories(p.getParent());

   491             } catch (IOException ex) {

   492                 throw new UncheckedIOException(ex);

   493             }

   494 

   495             final HttpResponse.BodyHandler<Path> proc =

   496                  HttpResponse.BodyHandler.asFile(p);

   497 

   498             return Optional.of(proc);

   499         }

   500     }

   501 

   502     /**

   503      * Currently this consumes all of the data and ignores it

   504      */

   505     static class NullSubscriber<T> implements HttpResponse.BodySubscriber<T> {

   506 

   507         volatile Flow.Subscription subscription;

   508         final CompletableFuture<T> cf = new MinimalFuture<>();

   509         final Optional<T> result;

   510 

   511         NullSubscriber(Optional<T> result) {

   512             this.result = result;

   513         }

   514 

   515         @Override

   516         public void onSubscribe(Flow.Subscription subscription) {

   517             this.subscription = subscription;

   518             subscription.request(Long.MAX_VALUE);

   519         }

   520 

   521         @Override

   522         public void onNext(List<ByteBuffer> items) {

   523             // NO-OP

   524         }

   525 

   526         @Override

   527         public void onError(Throwable throwable) {

   528             cf.completeExceptionally(throwable);

   529         }

   530 

   531         @Override

   532         public void onComplete() {

   533             if (result.isPresent()) {

   534                 cf.complete(result.get());

   535             } else {

   536                 cf.complete(null);

   537             }

   538         }

   539 

   540         @Override

   541         public CompletionStage<T> getBody() {

   542             return cf;

   543         }

   544     }

   545 }