1 /*

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

    29 import java.io.IOException;

    30 import java.io.InputStream;

    31 import java.io.InputStreamReader;

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

    33 import java.nio.ByteBuffer;

    34 import java.nio.channels.FileChannel;

    35 import java.nio.charset.Charset;

    36 import java.nio.charset.StandardCharsets;

    37 import java.nio.file.OpenOption;

    38 import java.nio.file.Path;

    39 import java.security.AccessControlContext;

    40 import java.security.AccessController;

    41 import java.security.PrivilegedActionException;

    42 import java.security.PrivilegedExceptionAction;

    43 import java.util.ArrayList;

    44 import java.util.Iterator;

    45 import java.util.List;

    46 import java.util.Objects;

    47 import java.util.Optional;

    48 import java.util.concurrent.ArrayBlockingQueue;

    49 import java.util.concurrent.BlockingQueue;

    50 import java.util.concurrent.CompletableFuture;

    51 import java.util.concurrent.CompletionStage;

    52 import java.util.concurrent.ConcurrentHashMap;

    53 import java.util.concurrent.Flow;

    54 import java.util.concurrent.Flow.Subscriber;

    55 import java.util.concurrent.Flow.Subscription;

    56 import java.util.concurrent.atomic.AtomicBoolean;

    57 import java.util.function.Consumer;

    58 import java.util.function.Function;

    59 import java.util.stream.Stream;

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

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

    62 

    63 class ResponseSubscribers {

    64 

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

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

    67         private Flow.Subscription subscription;

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

    69         private final AtomicBoolean subscribed = new AtomicBoolean();

    70 

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

    72             this.consumer = Objects.requireNonNull(consumer);

    73         }

    74 

    75         @Override

    76         public CompletionStage<Void> getBody() {

    77             return result;

    78         }

    79 

    80         @Override

    81         public void onSubscribe(Flow.Subscription subscription) {

    82             if (!subscribed.compareAndSet(false, true)) {

    83                 subscription.cancel();

    84             } else {

    85                 this.subscription = subscription;

    86                 subscription.request(1);

    87             }

    88         }

    89 

    90         @Override

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

    92             for (ByteBuffer item : items) {

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

    94                 item.get(buf);

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

    96             }

    97             subscription.request(1);

    98         }

    99 

   100         @Override

   101         public void onError(Throwable throwable) {

   102             result.completeExceptionally(throwable);

   103         }

   104 

   105         @Override

   106         public void onComplete() {

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

   108             result.complete(null);

   109         }

   110 

   111     }

   112 

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

   114 

   115         private final Path file;

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

   117 

   118         private volatile Flow.Subscription subscription;

   119         private volatile FileChannel out;

   120         private volatile AccessControlContext acc;

   121         private final OpenOption[] options;

   122 

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

   124             this.file = file;

   125             this.options = options;

   126         }

   127 

   128         void setAccessControlContext(AccessControlContext acc) {

   129             this.acc = acc;

   130         }

   131 

   132         @Override

   133         public void onSubscribe(Flow.Subscription subscription) {

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

   135                 throw new InternalError(

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

   137 

   138             this.subscription = subscription;

   139             try {

   140                 PrivilegedExceptionAction<FileChannel> pa =

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

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

   143             } catch (PrivilegedActionException pae) {

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

   145                 result.completeExceptionally(t);

   146                 subscription.cancel();

   147                 return;

   148             }

   149             subscription.request(1);

   150         }

   151 

   152         @Override

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

   154             try {

   155                 out.write(items.toArray(Utils.EMPTY_BB_ARRAY));

   156             } catch (IOException ex) {

   157                 Utils.close(out);

   158                 subscription.cancel();

   159                 result.completeExceptionally(ex);

   160             }

   161             subscription.request(1);

   162         }

   163 

   164         @Override

   165         public void onError(Throwable e) {

   166             result.completeExceptionally(e);

   167             Utils.close(out);

   168         }

   169 

   170         @Override

   171         public void onComplete() {

   172             Utils.close(out);

   173             result.complete(file);

   174         }

   175 

   176         @Override

   177         public CompletionStage<Path> getBody() {

   178             return result;

   179         }

   180     }

   181 

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

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

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

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

   186 

   187         private volatile Flow.Subscription subscription;

   188 

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

   190             this.finisher = finisher;

   191         }

   192 

   193         @Override

   194         public void onSubscribe(Flow.Subscription subscription) {

   195             if (this.subscription != null) {

   196                 subscription.cancel();

   197                 return;

   198             }

   199             this.subscription = subscription;

   200             // We can handle whatever you've got

   201             subscription.request(Long.MAX_VALUE);

   202         }

   203 

   204         @Override

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

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

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

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

   209             assert Utils.hasRemaining(items);

   210             received.addAll(items);

   211         }

   212 

   213         @Override

   214         public void onError(Throwable throwable) {

   215             received.clear();

   216             result.completeExceptionally(throwable);

   217         }

   218 

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

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

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

   222             int from = 0;

   223             for (ByteBuffer b : bytes) {

   224                 int l = b.remaining();

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

   226                 from += l;

   227             }

   228             return res;

   229         }

   230 

   231         @Override

   232         public void onComplete() {

   233             try {

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

   235                 received.clear();

   236             } catch (IllegalArgumentException e) {

   237                 result.completeExceptionally(e);

   238             }

   239         }

   240 

   241         @Override

   242         public CompletionStage<T> getBody() {

   243             return result;

   244         }

   245     }

   246 

   247     /**

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

   249      */

   250     static class HttpResponseInputStream extends InputStream

   251         implements HttpResponse.BodySubscriber<InputStream>

   252     {

   253         final static boolean DEBUG = Utils.DEBUG;

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

   255 

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

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

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

   259         private static final System.Logger DEBUG_LOGGER =

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

   261 

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

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

   264         private volatile Flow.Subscription subscription;

   265         private volatile boolean closed;

   266         private volatile Throwable failed;

   267         private volatile Iterator<ByteBuffer> currentListItr;

   268         private volatile ByteBuffer currentBuffer;

   269         private final AtomicBoolean subscribed = new AtomicBoolean();

   270 

   271         HttpResponseInputStream() {

   272             this(MAX_BUFFERS_IN_QUEUE);

   273         }

   274 

   275         HttpResponseInputStream(int maxBuffers) {

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

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

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

   279         }

   280 

   281         @Override

   282         public CompletionStage<InputStream> getBody() {

   283             // Returns the stream immediately, before the

   284             // response body is received.

   285             // This makes it possible for sendAsync().get().body()

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

   287             return CompletableFuture.completedStage(this);

   288         }

   289 

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

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

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

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

   294         // end of the flow has been reached.

   295         private ByteBuffer current() throws IOException {

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

   297                 // Check whether the stream is closed or exhausted

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

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

   300                 }

   301                 if (currentBuffer == LAST_BUFFER) break;

   302 

   303                 try {

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

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

   306                         // if none is available yet...

   307 

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

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

   310                         currentListItr = lb.iterator();

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

   312 

   313                         // Check whether an exception was encountered upstream

   314                         if (closed || failed != null)

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

   316 

   317                         // Check whether we're done.

   318                         if (lb == LAST_LIST) {

   319                             currentListItr = null;

   320                             currentBuffer = LAST_BUFFER;

   321                             break;

   322                         }

   323 

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

   325                         Flow.Subscription s = subscription;

   326                         if (s != null) {

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

   328                             s.request(1);

   329                         }

   330                         assert currentListItr != null;

   331                         if (lb.isEmpty()) continue;

   332                     }

   333                     assert currentListItr != null;

   334                     assert currentListItr.hasNext();

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

   336                     currentBuffer = currentListItr.next();

   337                 } catch (InterruptedException ex) {

   338                     // continue

   339                 }

   340             }

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

   342             return currentBuffer;

   343         }

   344 

   345         @Override

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

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

   348             // none is available

   349             ByteBuffer buffer;

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

   351 

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

   353             // in the current buffer.

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

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

   356 

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

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

   359             return read;

   360         }

   361 

   362         @Override

   363         public int read() throws IOException {

   364             ByteBuffer buffer;

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

   366             return buffer.get() & 0xFF;

   367         }

   368 

   369         @Override

   370         public void onSubscribe(Flow.Subscription s) {

   371             try {

   372                 if (!subscribed.compareAndSet(false, true)) {

   373                     s.cancel();

   374                 } else {

   375                     // check whether the stream is already closed.

   376                     // if so, we should cancel the subscription

   377                     // immediately.

   378                     boolean closed;

   379                     synchronized (this) {

   380                         closed = this.closed;

   381                         if (!closed) {

   382                             this.subscription = s;

   383                         }

   384                     }

   385                     if (closed) {

   386                         s.cancel();

   387                         return;

   388                     }

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

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

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

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

   393                 }

   394             } catch (Throwable t) {

   395                 failed = t;

   396                 try {

   397                     close();

   398                 } catch (IOException x) {

   399                     // OK

   400                 } finally {

   401                     onError(t);

   402                 }

   403             }

   404         }

   405 

   406         @Override

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

   408             Objects.requireNonNull(t);

   409             try {

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

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

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

   413                 }

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

   415             } catch (Throwable ex) {

   416                 failed = ex;

   417                 try {

   418                     close();

   419                 } catch (IOException ex1) {

   420                     // OK

   421                 } finally {

   422                     onError(ex);

   423                 }

   424             }

   425         }

   426 

   427         @Override

   428         public void onError(Throwable thrwbl) {

   429             subscription = null;

   430             failed = Objects.requireNonNull(thrwbl);

   431             // The client process that reads the input stream might

   432             // be blocked in queue.take().

   433             // Tries to offer LAST_LIST to the queue. If the queue is

   434             // full we don't care if we can't insert this buffer, as

   435             // the client can't be blocked in queue.take() in that case.

   436             // Adding LAST_LIST to the queue is harmless, as the client

   437             // should find failed != null before handling LAST_LIST.

   438             buffers.offer(LAST_LIST);

   439         }

   440 

   441         @Override

   442         public void onComplete() {

   443             subscription = null;

   444             onNext(LAST_LIST);

   445         }

   446 

   447         @Override

   448         public void close() throws IOException {

   449             Flow.Subscription s;

   450             synchronized (this) {

   451                 if (closed) return;

   452                 closed = true;

   453                 s = subscription;

   454                 subscription = null;

   455             }

   456             // s will be null if already completed

   457             try {

   458                 if (s != null) {

   459                     s.cancel();

   460                 }

   461             } finally {

   462                 buffers.offer(LAST_LIST);

   463                 super.close();

   464             }

   465         }

   466 

   467     }

   468 

   469     /**

   470      * A {@code Stream<String>} built on top of the Flow API.

   471      */

   472     static final class HttpLineStream implements HttpResponse.BodySubscriber<Stream<String>> {

   473 

   474         private final HttpResponseInputStream responseInputStream;

   475         private final Charset charset;

   476         private HttpLineStream(Charset charset) {

   477             this.charset = Objects.requireNonNull(charset);

   478             responseInputStream = new HttpResponseInputStream();

   479         }

   480 

   481         @Override

   482         public CompletionStage<Stream<String>> getBody() {

   483             return responseInputStream.getBody().thenApply((is) ->

   484                     new BufferedReader(new InputStreamReader(is, charset))

   485                             .lines().onClose(this::close));

   486         }

   487 

   488         @Override

   489         public void onSubscribe(Subscription subscription) {

   490             responseInputStream.onSubscribe(subscription);

   491         }

   492 

   493         @Override

   494         public void onNext(List<ByteBuffer> item) {

   495             responseInputStream.onNext(item);

   496         }

   497 

   498         @Override

   499         public void onError(Throwable throwable) {

   500             responseInputStream.onError(throwable);

   501         }

   502 

   503         @Override

   504         public void onComplete() {

   505             responseInputStream.onComplete();

   506         }

   507 

   508         void close() {

   509             try {

   510                 responseInputStream.close();

   511             } catch (IOException x) {

   512                 // ignore

   513             }

   514         }

   515 

   516         static HttpLineStream create(Charset charset) {

   517             return new HttpLineStream(Optional.ofNullable(charset).orElse(StandardCharsets.UTF_8));

   518         }

   519     }

   520 

   521     /**

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

   523      */

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

   525 

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

   527         private final Optional<T> result;

   528         private final AtomicBoolean subscribed = new AtomicBoolean();

   529 

   530         NullSubscriber(Optional<T> result) {

   531             this.result = result;

   532         }

   533 

   534         @Override

   535         public void onSubscribe(Flow.Subscription subscription) {

   536             if (!subscribed.compareAndSet(false, true)) {

   537                 subscription.cancel();

   538             } else {

   539                 subscription.request(Long.MAX_VALUE);

   540             }

   541         }

   542 

   543         @Override

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

   545             Objects.requireNonNull(items);

   546         }

   547 

   548         @Override

   549         public void onError(Throwable throwable) {

   550             cf.completeExceptionally(throwable);

   551         }

   552 

   553         @Override

   554         public void onComplete() {

   555             if (result.isPresent()) {

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

   557             } else {

   558                 cf.complete(null);

   559             }

   560         }

   561 

   562         @Override

   563         public CompletionStage<T> getBody() {

   564             return cf;

   565         }

   566     }

   567 

   568     /** An adapter between {@code BodySubscriber} and {@code Flow.Subscriber}. */

   569     static final class SubscriberAdapter<S extends Subscriber<? super List<ByteBuffer>>,R>

   570         implements HttpResponse.BodySubscriber<R>

   571     {

   572         private final CompletableFuture<R> cf = new MinimalFuture<>();

   573         private final S subscriber;

   574         private final Function<S,R> finisher;

   575         private volatile Subscription subscription;

   576 

   577         SubscriberAdapter(S subscriber, Function<S,R> finisher) {

   578             this.subscriber = Objects.requireNonNull(subscriber);

   579             this.finisher = Objects.requireNonNull(finisher);

   580         }

   581 

   582         @Override

   583         public void onSubscribe(Subscription subscription) {

   584             Objects.requireNonNull(subscription);

   585             if (this.subscription != null) {

   586                 subscription.cancel();

   587             } else {

   588                 this.subscription = subscription;

   589                 subscriber.onSubscribe(subscription);

   590             }

   591         }

   592 

   593         @Override

   594         public void onNext(List<ByteBuffer> item) {

   595             Objects.requireNonNull(item);

   596             try {

   597                 subscriber.onNext(item);

   598             } catch (Throwable throwable) {

   599                 subscription.cancel();

   600                 onError(throwable);

   601             }

   602         }

   603 

   604         @Override

   605         public void onError(Throwable throwable) {

   606             Objects.requireNonNull(throwable);

   607             try {

   608                 subscriber.onError(throwable);

   609             } finally {

   610                 cf.completeExceptionally(throwable);

   611             }

   612         }

   613 

   614         @Override

   615         public void onComplete() {

   616             try {

   617                 subscriber.onComplete();

   618             } finally {

   619                 try {

   620                     cf.complete(finisher.apply(subscriber));

   621                 } catch (Throwable throwable) {

   622                     cf.completeExceptionally(throwable);

   623                 }

   624             }

   625         }

   626 

   627         @Override

   628         public CompletionStage<R> getBody() {

   629             return cf;

   630         }

   631     }

   632 }