1 /*

     2  * Copyright (c) 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.

     8  *

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

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

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

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

    13  * accompanied this code).

    14  *

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

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

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

    18  *

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

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

    21  * questions.

    22  */

    23 

    24 import jdk.incubator.http.HttpResponse;

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

    26 

    27 import java.io.IOException;

    28 import java.io.InputStream;

    29 import java.io.UncheckedIOException;

    30 import java.nio.ByteBuffer;

    31 import java.util.List;

    32 import java.util.concurrent.CompletableFuture;

    33 import java.util.concurrent.CompletionException;

    34 import java.util.concurrent.CompletionStage;

    35 import java.util.concurrent.CountDownLatch;

    36 import java.util.concurrent.ExecutionException;

    37 import java.util.concurrent.Flow;

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

    39 import java.util.concurrent.atomic.AtomicLong;

    40 

    41 import org.testng.annotations.Test;

    42 

    43 /*

    44  * @test

    45  * @summary Simple smoke test for BodySubscriber.asInputStream();

    46  * @run testng/othervm HttpResponseInputStreamTest

    47  * @author daniel fuchs

    48  */

    49 public class HttpResponseInputStreamTest {

    50 

    51     static class TestException extends IOException {}

    52 

    53     public static void main(String[] args) throws InterruptedException, ExecutionException {

    54         testOnError();

    55     }

    56 

    57     /**

    58      * Tests that a client will be unblocked and will throw an IOException

    59      * if an error occurs while the client is waiting for more data.

    60      * @throws InterruptedException

    61      * @throws ExecutionException

    62      */

    63     @Test

    64     public static void testOnError() throws InterruptedException, ExecutionException {

    65         CountDownLatch latch = new CountDownLatch(1);

    66         BodySubscriber<InputStream> isb = BodySubscriber.asInputStream();

    67         ErrorTestSubscription s = new ErrorTestSubscription(isb);

    68         CompletionStage<Throwable> cs =

    69                 isb.getBody().thenApplyAsync((is) -> s.accept(latch, is));

    70         latch.await();

    71         isb.onSubscribe(s);

    72         s.t.join();

    73         Throwable result = cs.toCompletableFuture().get();

    74         Throwable t = result;

    75         if (!(t instanceof IOException)) {

    76             throw new RuntimeException("Failed to receive excpected IOException", result);

    77         } else {

    78             System.out.println("Got expected exception: " + t);

    79         }

    80         while (t != null) {

    81             if (t instanceof TestException) break;

    82             t = t.getCause();

    83         }

    84         if (t instanceof TestException) {

    85             System.out.println("Got expected cause: " + t);

    86         } else {

    87             throw new RuntimeException("Failed to receive excpected TestException", result);

    88         }

    89     }

    90 

    91     static class ErrorTestSubscription implements Flow.Subscription {

    92         final BodySubscriber<InputStream> isb;

    93         final Thread t = new Thread() {

    94             @Override

    95             public void run() {

    96                 try {

    97                     // Give time to

    98                     System.out.println("waiting...");

    99                     Thread.sleep(1000);

   100                 } catch (InterruptedException e) {

   101 

   102                 }

   103                 System.out.println("Calling onError...");

   104                 isb.onError(new TestException());

   105             }

   106         };

   107 

   108         ErrorTestSubscription(BodySubscriber<InputStream> isb) {

   109             this.isb = isb;

   110         }

   111 

   112         int requested = 0;

   113 

   114         @Override

   115         public void request(long n) {

   116             System.out.println("Got request: " + n);

   117             if (requested == 0 && n > 0) {

   118                 //isb.onNext(List.of(java.nio.ByteBuffer.wrap(new byte[] {0x01})));

   119                 requested += n;

   120                 t.start();

   121             }

   122         }

   123 

   124         @Override

   125         public void cancel() {

   126         }

   127 

   128         public Throwable accept(CountDownLatch latch, InputStream is) {

   129             System.out.println("got " + is);

   130             try {

   131                 latch.countDown();

   132                 System.out.println("reading all bytes");

   133                 is.readAllBytes();

   134                 System.out.println("all bytes read");

   135             } catch (IOException e) {

   136                 return e;

   137             } finally {

   138                 try {

   139                     is.close();

   140                 } catch (IOException e) {

   141                     return e;

   142                 }

   143             }

   144             return is == null ? new NullPointerException() : null;

   145         }

   146     }

   147 

   148     static InputStream close(InputStream is) {

   149         try {

   150             is.close();

   151         } catch (IOException io) {

   152             throw new CompletionException(io);

   153         }

   154         return is;

   155     }

   156 

   157     @Test

   158     public static void testCloseAndSubscribe()

   159             throws InterruptedException, ExecutionException

   160     {

   161         BodySubscriber<InputStream> isb = BodySubscriber.asInputStream();

   162         TestCancelOnCloseSubscription s = new TestCancelOnCloseSubscription();

   163         InputStream is = isb.getBody()

   164                 .thenApply(HttpResponseInputStreamTest::close)

   165                 .toCompletableFuture()

   166                 .get();

   167         isb.onSubscribe(s);

   168         System.out.println(s);

   169         if (!s.cancelled.get()) {

   170             throw new RuntimeException("subscription not cancelled");

   171         }

   172         if (s.request.get() > 0) {

   173             throw new RuntimeException("subscription has demand");

   174         }

   175     }

   176 

   177     static byte[] readAllBytes(InputStream is) {

   178         try {

   179             return is.readAllBytes();

   180         } catch (IOException io) {

   181             io.printStackTrace();

   182             throw new CompletionException(io);

   183         }

   184     }

   185 

   186     @Test

   187     public static void testSubscribeAndClose()

   188             throws InterruptedException, ExecutionException

   189     {

   190         BodySubscriber<InputStream> isb = BodySubscriber.asInputStream();

   191         TestCancelOnCloseSubscription s = new TestCancelOnCloseSubscription();

   192         InputStream is = isb.getBody().toCompletableFuture().get();

   193         isb.onSubscribe(s);

   194         if (s.cancelled.get()) {

   195             throw new RuntimeException("subscription cancelled");

   196         }

   197         CompletableFuture<String> cf = CompletableFuture.supplyAsync(

   198                 () -> HttpResponseInputStreamTest.readAllBytes(is))

   199                 .thenApply(String::new);

   200         while (s.request.get() == 0) {

   201             Thread.sleep(100);

   202         }

   203         isb.onNext(List.of(ByteBuffer.wrap("coucou".getBytes())));

   204         close(is);

   205         System.out.println(s);

   206         if (!s.cancelled.get()) {

   207             throw new RuntimeException("subscription not cancelled");

   208         }

   209         if (s.request.get() == 0) {

   210             throw new RuntimeException("subscription has no demand");

   211         }

   212         try {

   213             System.out.println("read " + cf.get() + "!");

   214             throw new RuntimeException("expected IOException not raised");

   215         } catch (ExecutionException | CompletionException x) {

   216             if (x.getCause() instanceof IOException) {

   217                 System.out.println("Got expected IOException: " + x.getCause());

   218             } else {

   219                 throw x;

   220             }

   221         }

   222     }

   223 

   224     static class TestCancelOnCloseSubscription implements Flow.Subscription {

   225         final AtomicLong request = new AtomicLong();

   226         final AtomicBoolean cancelled = new AtomicBoolean();

   227 

   228         @Override

   229         public void request(long n) {

   230             request.addAndGet(n);

   231         }

   232 

   233         @Override

   234         public void cancel() {

   235             cancelled.set(true);

   236         }

   237 

   238         @Override

   239         public String toString() {

   240             return String.format("%s(request=%d, cancelled=%s)",

   241                     this.getClass().getSimpleName(),

   242                     request.get(),

   243                     cancelled.get());

   244         }

   245     }

   246 }