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.internal;

    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.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.Objects;

    46 import java.util.Optional;

    47 import java.util.concurrent.ArrayBlockingQueue;

    48 import java.util.concurrent.BlockingQueue;

    49 import java.util.concurrent.CompletableFuture;

    50 import java.util.concurrent.CompletionStage;

    51 import java.util.concurrent.ConcurrentHashMap;

    52 import java.util.concurrent.Flow;

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

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

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

    56 import java.util.function.Consumer;

    57 import java.util.function.Function;

    58 import java.util.stream.Stream;

    59 import jdk.incubator.http.HttpResponse.BodySubscriber;

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

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

    62 import static java.nio.charset.StandardCharsets.UTF_8;

    63 

    64 public class ResponseSubscribers {

    65 

    66     public static class ConsumerSubscriber implements BodySubscriber<Void> {

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

    68         private Flow.Subscription subscription;

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

    70         private final AtomicBoolean subscribed = new AtomicBoolean();

    71 

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

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

    74         }

    75 

    76         @Override

    77         public CompletionStage<Void> getBody() {

    78             return result;

    79         }

    80 

    81         @Override

    82         public void onSubscribe(Flow.Subscription subscription) {

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

    84                 subscription.cancel();

    85             } else {

    86                 this.subscription = subscription;

    87                 subscription.request(1);

    88             }

    89         }

    90 

    91         @Override

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

    93             for (ByteBuffer item : items) {

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

    95                 item.get(buf);

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

    97             }

    98             subscription.request(1);

    99         }

   100 

   101         @Override

   102         public void onError(Throwable throwable) {

   103             result.completeExceptionally(throwable);

   104         }

   105 

   106         @Override

   107         public void onComplete() {

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

   109             result.complete(null);

   110         }

   111 

   112     }

   113 

   114     public static class PathSubscriber implements BodySubscriber<Path> {

   115 

   116         private final Path file;

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

   118 

   119         private volatile Flow.Subscription subscription;

   120         private volatile FileChannel out;

   121         private volatile AccessControlContext acc;

   122         private final OpenOption[] options;

   123 

   124         public PathSubscriber(Path file, OpenOption... options) {

   125             this.file = file;

   126             this.options = options;

   127         }

   128 

   129         public void setAccessControlContext(AccessControlContext acc) {

   130             this.acc = acc;

   131         }

   132 

   133         @Override

   134         public void onSubscribe(Flow.Subscription subscription) {

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

   136                 throw new InternalError(

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

   138 

   139             this.subscription = subscription;

   140             try {

   141                 PrivilegedExceptionAction<FileChannel> pa =

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

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

   144             } catch (PrivilegedActionException pae) {

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

   146                 result.completeExceptionally(t);

   147                 subscription.cancel();

   148                 return;

   149             }

   150             subscription.request(1);

   151         }

   152 

   153         @Override

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

   155             try {

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

   157             } catch (IOException ex) {

   158                 Utils.close(out);

   159                 subscription.cancel();

   160                 result.completeExceptionally(ex);

   161             }

   162             subscription.request(1);

   163         }

   164 

   165         @Override

   166         public void onError(Throwable e) {

   167             result.completeExceptionally(e);

   168             Utils.close(out);

   169         }

   170 

   171         @Override

   172         public void onComplete() {

   173             Utils.close(out);

   174             result.complete(file);

   175         }

   176 

   177         @Override

   178         public CompletionStage<Path> getBody() {

   179             return result;

   180         }

   181     }

   182 

   183     public static class ByteArraySubscriber<T> implements BodySubscriber<T> {

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

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

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

   187 

   188         private volatile Flow.Subscription subscription;

   189 

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

   191             this.finisher = finisher;

   192         }

   193 

   194         @Override

   195         public void onSubscribe(Flow.Subscription subscription) {

   196             if (this.subscription != null) {

   197                 subscription.cancel();

   198                 return;

   199             }

   200             this.subscription = subscription;

   201             // We can handle whatever you've got

   202             subscription.request(Long.MAX_VALUE);

   203         }

   204 

   205         @Override

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

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

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

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

   210             assert Utils.hasRemaining(items);

   211             received.addAll(items);

   212         }

   213 

   214         @Override

   215         public void onError(Throwable throwable) {

   216             received.clear();

   217             result.completeExceptionally(throwable);

   218         }

   219 

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

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

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

   223             int from = 0;

   224             for (ByteBuffer b : bytes) {

   225                 int l = b.remaining();

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

   227                 from += l;

   228             }

   229             return res;

   230         }

   231 

   232         @Override

   233         public void onComplete() {

   234             try {

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

   236                 received.clear();

   237             } catch (IllegalArgumentException e) {

   238                 result.completeExceptionally(e);

   239             }

   240         }

   241 

   242         @Override

   243         public CompletionStage<T> getBody() {

   244             return result;

   245         }

   246     }

   247 

   248     /**

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

   250      */

   251     public static class HttpResponseInputStream extends InputStream

   252         implements BodySubscriber<InputStream>

   253     {

   254         final static boolean DEBUG = Utils.DEBUG;

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

   256 

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

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

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

   260         private static final System.Logger DEBUG_LOGGER =

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

   262 

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

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

   265         private volatile Flow.Subscription subscription;

   266         private volatile boolean closed;

   267         private volatile Throwable failed;

   268         private volatile Iterator<ByteBuffer> currentListItr;

   269         private volatile ByteBuffer currentBuffer;

   270         private final AtomicBoolean subscribed = new AtomicBoolean();

   271 

   272         public HttpResponseInputStream() {

   273             this(MAX_BUFFERS_IN_QUEUE);

   274         }

   275 

   276         HttpResponseInputStream(int maxBuffers) {

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

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

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

   280         }

   281 

   282         @Override

   283         public CompletionStage<InputStream> getBody() {

   284             // Returns the stream immediately, before the

   285             // response body is received.

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

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

   288             return CompletableFuture.completedStage(this);

   289         }

   290 

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

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

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

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

   295         // end of the flow has been reached.

   296         private ByteBuffer current() throws IOException {

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

   298                 // Check whether the stream is closed or exhausted

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

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

   301                 }

   302                 if (currentBuffer == LAST_BUFFER) break;

   303 

   304                 try {

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

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

   307                         // if none is available yet...

   308 

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

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

   311                         currentListItr = lb.iterator();

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

   313 

   314                         // Check whether an exception was encountered upstream

   315                         if (closed || failed != null)

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

   317 

   318                         // Check whether we're done.

   319                         if (lb == LAST_LIST) {

   320                             currentListItr = null;

   321                             currentBuffer = LAST_BUFFER;

   322                             break;

   323                         }

   324 

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

   326                         Flow.Subscription s = subscription;

   327                         if (s != null) {

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

   329                             s.request(1);

   330                         }

   331                         assert currentListItr != null;

   332                         if (lb.isEmpty()) continue;

   333                     }

   334                     assert currentListItr != null;

   335                     assert currentListItr.hasNext();

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

   337                     currentBuffer = currentListItr.next();

   338                 } catch (InterruptedException ex) {

   339                     // continue

   340                 }

   341             }

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

   343             return currentBuffer;

   344         }

   345 

   346         @Override

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

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

   349             // none is available

   350             ByteBuffer buffer;

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

   352 

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

   354             // in the current buffer.

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

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

   357 

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

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

   360             return read;

   361         }

   362 

   363         @Override

   364         public int read() throws IOException {

   365             ByteBuffer buffer;

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

   367             return buffer.get() & 0xFF;

   368         }

   369 

   370         @Override

   371         public void onSubscribe(Flow.Subscription s) {

   372             try {

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

   374                     s.cancel();

   375                 } else {

   376                     // check whether the stream is already closed.

   377                     // if so, we should cancel the subscription

   378                     // immediately.

   379                     boolean closed;

   380                     synchronized (this) {

   381                         closed = this.closed;

   382                         if (!closed) {

   383                             this.subscription = s;

   384                         }

   385                     }

   386                     if (closed) {

   387                         s.cancel();

   388                         return;

   389                     }

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

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

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

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

   394                 }

   395             } catch (Throwable t) {

   396                 failed = t;

   397                 try {

   398                     close();

   399                 } catch (IOException x) {

   400                     // OK

   401                 } finally {

   402                     onError(t);

   403                 }

   404             }

   405         }

   406 

   407         @Override

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

   409             Objects.requireNonNull(t);

   410             try {

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

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

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

   414                 }

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

   416             } catch (Throwable ex) {

   417                 failed = ex;

   418                 try {

   419                     close();

   420                 } catch (IOException ex1) {

   421                     // OK

   422                 } finally {

   423                     onError(ex);

   424                 }

   425             }

   426         }

   427 

   428         @Override

   429         public void onError(Throwable thrwbl) {

   430             subscription = null;

   431             failed = Objects.requireNonNull(thrwbl);

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

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

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

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

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

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

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

   439             buffers.offer(LAST_LIST);

   440         }

   441 

   442         @Override

   443         public void onComplete() {

   444             subscription = null;

   445             onNext(LAST_LIST);

   446         }

   447 

   448         @Override

   449         public void close() throws IOException {

   450             Flow.Subscription s;

   451             synchronized (this) {

   452                 if (closed) return;

   453                 closed = true;

   454                 s = subscription;

   455                 subscription = null;

   456             }

   457             // s will be null if already completed

   458             try {

   459                 if (s != null) {

   460                     s.cancel();

   461                 }

   462             } finally {

   463                 buffers.offer(LAST_LIST);

   464                 super.close();

   465             }

   466         }

   467 

   468     }

   469 

   470     /**

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

   472      */

   473     public static final class HttpLineStream implements BodySubscriber<Stream<String>> {

   474 

   475         private final HttpResponseInputStream responseInputStream;

   476         private final Charset charset;

   477         private HttpLineStream(Charset charset) {

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

   479             responseInputStream = new HttpResponseInputStream();

   480         }

   481 

   482         @Override

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

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

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

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

   487         }

   488 

   489         @Override

   490         public void onSubscribe(Subscription subscription) {

   491             responseInputStream.onSubscribe(subscription);

   492         }

   493 

   494         @Override

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

   496             responseInputStream.onNext(item);

   497         }

   498 

   499         @Override

   500         public void onError(Throwable throwable) {

   501             responseInputStream.onError(throwable);

   502         }

   503 

   504         @Override

   505         public void onComplete() {

   506             responseInputStream.onComplete();

   507         }

   508 

   509         void close() {

   510             try {

   511                 responseInputStream.close();

   512             } catch (IOException x) {

   513                 // ignore

   514             }

   515         }

   516 

   517         public static HttpLineStream create(Charset charset) {

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

   519         }

   520     }

   521 

   522     /**

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

   524      */

   525     public static class NullSubscriber<T> implements BodySubscriber<T> {

   526 

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

   528         private final Optional<T> result;

   529         private final AtomicBoolean subscribed = new AtomicBoolean();

   530 

   531         public NullSubscriber(Optional<T> result) {

   532             this.result = result;

   533         }

   534 

   535         @Override

   536         public void onSubscribe(Flow.Subscription subscription) {

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

   538                 subscription.cancel();

   539             } else {

   540                 subscription.request(Long.MAX_VALUE);

   541             }

   542         }

   543 

   544         @Override

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

   546             Objects.requireNonNull(items);

   547         }

   548 

   549         @Override

   550         public void onError(Throwable throwable) {

   551             cf.completeExceptionally(throwable);

   552         }

   553 

   554         @Override

   555         public void onComplete() {

   556             if (result.isPresent()) {

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

   558             } else {

   559                 cf.complete(null);

   560             }

   561         }

   562 

   563         @Override

   564         public CompletionStage<T> getBody() {

   565             return cf;

   566         }

   567     }

   568 

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

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

   571         implements BodySubscriber<R>

   572     {

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

   574         private final S subscriber;

   575         private final Function<S,R> finisher;

   576         private volatile Subscription subscription;

   577 

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

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

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

   581         }

   582 

   583         @Override

   584         public void onSubscribe(Subscription subscription) {

   585             Objects.requireNonNull(subscription);

   586             if (this.subscription != null) {

   587                 subscription.cancel();

   588             } else {

   589                 this.subscription = subscription;

   590                 subscriber.onSubscribe(subscription);

   591             }

   592         }

   593 

   594         @Override

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

   596             Objects.requireNonNull(item);

   597             try {

   598                 subscriber.onNext(item);

   599             } catch (Throwable throwable) {

   600                 subscription.cancel();

   601                 onError(throwable);

   602             }

   603         }

   604 

   605         @Override

   606         public void onError(Throwable throwable) {

   607             Objects.requireNonNull(throwable);

   608             try {

   609                 subscriber.onError(throwable);

   610             } finally {

   611                 cf.completeExceptionally(throwable);

   612             }

   613         }

   614 

   615         @Override

   616         public void onComplete() {

   617             try {

   618                 subscriber.onComplete();

   619             } finally {

   620                 try {

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

   622                 } catch (Throwable throwable) {

   623                     cf.completeExceptionally(throwable);

   624                 }

   625             }

   626         }

   627 

   628         @Override

   629         public CompletionStage<R> getBody() {

   630             return cf;

   631         }

   632     }

   633 }