1 /*

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

     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 /*

    25  * @test

    26  * @summary Tests what happens when response body handlers and subscribers

    27  *          throw unexpected exceptions.

    28  * @library /lib/testlibrary http2/server

    29  * @build jdk.testlibrary.SimpleSSLContext HttpServerAdapters ThrowingSubscribers

    30  * @modules java.base/sun.net.www.http

    31  *          java.net.http/jdk.internal.net.http.common

    32  *          java.net.http/jdk.internal.net.http.frame

    33  *          java.net.http/jdk.internal.net.http.hpack

    34  * @run testng/othervm ThrowingSubscribers

    35  */

    36 

    37 import com.sun.net.httpserver.HttpExchange;

    38 import com.sun.net.httpserver.HttpHandler;

    39 import com.sun.net.httpserver.HttpServer;

    40 import com.sun.net.httpserver.HttpsConfigurator;

    41 import com.sun.net.httpserver.HttpsServer;

    42 import jdk.testlibrary.SimpleSSLContext;

    43 import org.testng.annotations.AfterTest;

    44 import org.testng.annotations.AfterClass;

    45 import org.testng.annotations.BeforeTest;

    46 import org.testng.annotations.DataProvider;

    47 import org.testng.annotations.Test;

    48 

    49 import javax.net.ssl.SSLContext;

    50 import java.io.BufferedReader;

    51 import java.io.IOException;

    52 import java.io.InputStream;

    53 import java.io.InputStreamReader;

    54 import java.io.OutputStream;

    55 import java.net.InetSocketAddress;

    56 import java.net.URI;

    57 import java.net.http.HttpClient;

    58 import java.net.http.HttpHeaders;

    59 import java.net.http.HttpRequest;

    60 import java.net.http.HttpResponse;

    61 import java.net.http.HttpResponse.BodyHandler;

    62 import java.net.http.HttpResponse.BodySubscriber;

    63 import java.nio.ByteBuffer;

    64 import java.nio.charset.StandardCharsets;

    65 import java.util.List;

    66 import java.util.concurrent.CompletableFuture;

    67 import java.util.concurrent.CompletionStage;

    68 import java.util.concurrent.ConcurrentHashMap;

    69 import java.util.concurrent.ConcurrentMap;

    70 import java.util.concurrent.Executor;

    71 import java.util.concurrent.Executors;

    72 import java.util.concurrent.Flow;

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

    74 import java.util.function.BiFunction;

    75 import java.util.function.Consumer;

    76 import java.util.function.Predicate;

    77 import java.util.function.Supplier;

    78 import java.util.stream.Collectors;

    79 import java.util.stream.Stream;

    80 

    81 import static java.lang.System.out;

    82 import static java.lang.String.format;

    83 import static java.net.http.HttpResponse.BodySubscriber.asString;

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

    85 import static org.testng.Assert.assertEquals;

    86 import static org.testng.Assert.assertTrue;

    87 

    88 public class ThrowingSubscribers implements HttpServerAdapters {

    89 

    90     SSLContext sslContext;

    91     HttpTestServer httpTestServer;    // HTTP/1.1    [ 4 servers ]

    92     HttpTestServer httpsTestServer;   // HTTPS/1.1

    93     HttpTestServer http2TestServer;   // HTTP/2 ( h2c )

    94     HttpTestServer https2TestServer;  // HTTP/2 ( h2  )

    95     String httpURI_fixed;

    96     String httpURI_chunk;

    97     String httpsURI_fixed;

    98     String httpsURI_chunk;

    99     String http2URI_fixed;

   100     String http2URI_chunk;

   101     String https2URI_fixed;

   102     String https2URI_chunk;

   103 

   104     static final int ITERATION_COUNT = 2;

   105     // a shared executor helps reduce the amount of threads created by the test

   106     static final Executor executor = new TestExecutor(Executors.newCachedThreadPool());

   107     static final ConcurrentMap<String, Throwable> FAILURES = new ConcurrentHashMap<>();

   108     static volatile boolean tasksFailed;

   109 

   110     static class TestExecutor implements Executor {

   111         final AtomicLong tasks = new AtomicLong();

   112         Executor executor;

   113         TestExecutor(Executor executor) {

   114             this.executor = executor;

   115         }

   116 

   117         @Override

   118         public void execute(Runnable command) {

   119             long id = tasks.incrementAndGet();

   120             executor.execute(() -> {

   121                 try {

   122                     command.run();

   123                 } catch (Throwable t) {

   124                     tasksFailed = true;

   125                     System.out.printf("Task %s failed: %s%n", id, t);

   126                     System.err.printf("Task %s failed: %s%n", id, t);

   127                     FAILURES.putIfAbsent("Task " + id, t);

   128                     throw t;

   129                 }

   130             });

   131         }

   132     }

   133 

   134     @AfterClass

   135     static final void printFailedTests() {

   136         if (FAILURES.isEmpty()) return;

   137         out.println("Failed tests: ");

   138         FAILURES.entrySet().forEach((e) -> {

   139                 out.printf("\t%s: %s%n", e.getKey(), e.getValue());

   140                 e.getValue().printStackTrace();

   141         });

   142         if (tasksFailed) {

   143             throw new RuntimeException("Some tasks failed");

   144         }

   145     }

   146 

   147     @DataProvider(name = "variants")

   148     public Object[][] variants() {

   149         return new Object[][]{

   150                 { httpURI_fixed,    false },

   151                 { httpURI_chunk,    false },

   152                 { httpsURI_fixed,   false },

   153                 { httpsURI_chunk,   false },

   154                 { http2URI_fixed,   false },

   155                 { http2URI_chunk,   false },

   156                 { https2URI_fixed,  false },

   157                 { https2URI_chunk,  false },

   158 

   159                 { httpURI_fixed,    true },

   160                 { httpURI_chunk,    true },

   161                 { httpsURI_fixed,   true },

   162                 { httpsURI_chunk,   true },

   163                 { http2URI_fixed,   true },

   164                 { http2URI_chunk,   true },

   165                 { https2URI_fixed,  true },

   166                 { https2URI_chunk,  true },

   167         };

   168     }

   169 

   170     HttpClient newHttpClient() {

   171         return HttpClient.newBuilder()

   172                          .executor(executor)

   173                          .sslContext(sslContext)

   174                          .build();

   175     }

   176 

   177     @Test(dataProvider = "variants")

   178     public void testNoThrows(String uri, boolean sameClient) throws Exception {

   179         HttpClient client = null;

   180         for (int i=0; i< ITERATION_COUNT; i++) {

   181             if (!sameClient || client == null)

   182                 client = newHttpClient();

   183 

   184             HttpRequest req = HttpRequest.newBuilder(URI.create(uri))

   185                     .build();

   186             BodyHandler<String> handler = new ThrowingBodyHandler((w) -> {},

   187                     BodyHandler.asString());

   188             HttpResponse<String> response = client.send(req, handler);

   189             String body = response.body();

   190             assertEquals(URI.create(body).getPath(), URI.create(uri).getPath());

   191         }

   192     }

   193 

   194     @Test(dataProvider = "variants")

   195     public void testThrowingAsString(String uri, boolean sameClient) throws Exception {

   196         String test = format("testThrowingAsString(%s,%b)", uri, sameClient);

   197         testThrowing(test, uri, sameClient, BodyHandler::asString,

   198                 this::shouldHaveThrown, false);

   199     }

   200 

   201     @Test(dataProvider = "variants")

   202     public void testThrowingAsLines(String uri, boolean sameClient) throws Exception {

   203         String test =  format("testThrowingAsLines(%s,%b)", uri, sameClient);

   204         testThrowing(test, uri, sameClient, BodyHandler::asLines,

   205                 this::checkAsLines, false);

   206     }

   207 

   208     @Test(dataProvider = "variants")

   209     public void testThrowingAsInputStream(String uri, boolean sameClient) throws Exception {

   210         String test = format("testThrowingAsInputStream(%s,%b)", uri, sameClient);

   211         testThrowing(test, uri, sameClient, BodyHandler::asInputStream,

   212                 this::checkAsInputStream, false);

   213     }

   214 

   215     @Test(dataProvider = "variants")

   216     public void testThrowingAsStringAsync(String uri, boolean sameClient) throws Exception {

   217         String test = format("testThrowingAsStringAsync(%s,%b)", uri, sameClient);

   218         testThrowing(uri, sameClient, BodyHandler::asString,

   219                 this::shouldHaveThrown, true);

   220     }

   221 

   222     @Test(dataProvider = "variants")

   223     public void testThrowingAsLinesAsync(String uri, boolean sameClient) throws Exception {

   224         String test = format("testThrowingAsLinesAsync(%s,%b)", uri, sameClient);

   225         testThrowing(test, uri, sameClient, BodyHandler::asLines,

   226                 this::checkAsLines, true);

   227     }

   228 

   229     @Test(dataProvider = "variants")

   230     public void testThrowingAsInputStreamAsync(String uri, boolean sameClient) throws Exception {

   231         String test = format("testThrowingAsInputStreamAsync(%s,%b)", uri, sameClient);

   232         testThrowing(test, uri, sameClient, BodyHandler::asInputStream,

   233                 this::checkAsInputStream, true);

   234     }

   235 

   236     private <T,U> void testThrowing(String name, String uri, boolean sameClient,

   237                                     Supplier<BodyHandler<T>> handlers,

   238                                     Finisher finisher, boolean async)

   239             throws Exception {

   240         out.printf("%n%s%n", name);

   241         try {

   242             testThrowing(uri, sameClient, handlers, finisher, async);

   243         } catch (Error | Exception x) {

   244             FAILURES.putIfAbsent(name, x);

   245             throw x;

   246         }

   247     }

   248 

   249     private <T,U> void testThrowing(String uri, boolean sameClient,

   250         Supplier<BodyHandler<T>> handlers,

   251         Finisher finisher, boolean async)

   252             throws Exception {

   253         HttpClient client = null;

   254         RuntimeExceptionThrower thrower = new RuntimeExceptionThrower();

   255         for (Where where : Where.values()) {

   256             if (where == Where.ON_SUBSCRIBE) continue;

   257             if (where == Where.ON_ERROR) continue;

   258             if (where == Where.GET_BODY) continue; // doesn't work with HTTP/2

   259             if (!sameClient || client == null)

   260                 client = newHttpClient();

   261 

   262             HttpRequest req = HttpRequest.newBuilder(URI.create(uri))

   263                     .build();

   264             BodyHandler<T> handler = new ThrowingBodyHandler(where.select(thrower), handlers.get());

   265             System.out.println("try throwing in " + where);

   266             HttpResponse<T> response = null;

   267             if (async) {

   268                 try {

   269                     response = client.sendAsync(req, handler).join();

   270                 } catch (Error | Exception x) {

   271                     UncheckedCustomException cause = findCause(x,

   272                             UncheckedCustomException.class::isInstance);

   273                     if (cause == null) throw x;

   274                     System.out.println("Got expected exception: " + cause);

   275                 }

   276             } else {

   277                 try {

   278                     response = client.send(req, handler);

   279                 } catch (UncheckedCustomException t) {

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

   281                 }

   282             }

   283             if (response != null) {

   284                 finisher.finish(where, response);

   285             }

   286         }

   287     }

   288 

   289     enum Where {BODY_HANDLER, ON_SUBSCRIBE, ON_NEXT, ON_COMPLETE, ON_ERROR, GET_BODY;

   290         public Consumer<Where> select(Consumer<Where> consumer) {

   291             return new Consumer<Where>() {

   292                 @Override

   293                 public void accept(Where where) {

   294                     if (Where.this == where) {

   295                         consumer.accept(where);

   296                     }

   297                 }

   298             };

   299         }

   300     }

   301 

   302     interface Finisher<T,U> {

   303         U finish(Where w, HttpResponse<T> resp) throws IOException;

   304     }

   305 

   306     <T,U> U shouldHaveThrown(Where w, HttpResponse<T> resp) {

   307         throw new RuntimeException("Expected exception not thrown in " + w);

   308     }

   309 

   310     List<String> checkAsLines(Where w, HttpResponse<Stream<String>> resp) {

   311         switch(w) {

   312             case BODY_HANDLER: return shouldHaveThrown(w, resp);

   313             case ON_SUBSCRIBE: return shouldHaveThrown(w, resp);

   314             case GET_BODY: return shouldHaveThrown(w, resp);

   315             default: break;

   316         }

   317         List<String> result = null;

   318         try {

   319             result = resp.body().collect(Collectors.toList());

   320         } catch (Error | Exception x) {

   321             UncheckedCustomException cause =

   322                     findCause(x, UncheckedCustomException.class::isInstance);

   323             if (cause != null) {

   324                 out.println("Got expected exception in " + w + ": " + x);

   325                 return result;

   326             }

   327             throw x;

   328         }

   329         throw new RuntimeException("Expected exception not thrown in " + w);

   330     }

   331 

   332     List<String> checkAsInputStream(Where w, HttpResponse<InputStream> resp)

   333             throws IOException

   334     {

   335         switch(w) {

   336             case BODY_HANDLER: return shouldHaveThrown(w, resp);

   337             case ON_SUBSCRIBE: return shouldHaveThrown(w, resp);

   338             case GET_BODY: return shouldHaveThrown(w, resp);

   339             default: break;

   340         }

   341         List<String> result = null;

   342         try (InputStreamReader r1 = new InputStreamReader(resp.body(), UTF_8);

   343              BufferedReader r = new BufferedReader(r1)) {

   344             try {

   345                 result = r.lines().collect(Collectors.toList());

   346             } catch (Error | Exception x) {

   347                 UncheckedCustomException cause =

   348                         findCause(x, UncheckedCustomException.class::isInstance);

   349                 if (cause != null) {

   350                     out.println("Got expected exception in " + w + ": " + x);

   351                     return result;

   352                 }

   353                 throw x;

   354             }

   355         }

   356         throw new RuntimeException("Expected exception not thrown in " + w);

   357     }

   358 

   359     private static <E extends Throwable> E findCause(Throwable x,

   360                                                      Predicate<Throwable> filter) {

   361         while (x != null && !filter.test(x)) x = x.getCause();

   362         return (E)x;

   363     }

   364 

   365     static class RuntimeExceptionThrower implements Consumer<Where> {

   366         @Override

   367         public void accept(Where where) {

   368             throw new UncheckedCustomException(where.name());

   369         }

   370     }

   371 

   372     static class UncheckedCustomException extends RuntimeException {

   373         UncheckedCustomException(String message) {

   374             super(message);

   375         }

   376         UncheckedCustomException(String message, Throwable cause) {

   377             super(message, cause);

   378         }

   379     }

   380 

   381     static class ThrowingBodyHandler<T> implements BodyHandler<T> {

   382         final Consumer<Where> throwing;

   383         final BodyHandler<T> bodyHandler;

   384         ThrowingBodyHandler(Consumer<Where> throwing, BodyHandler<T> bodyHandler) {

   385             this.throwing = throwing;

   386             this.bodyHandler = bodyHandler;

   387         }

   388         @Override

   389         public BodySubscriber<T> apply(int statusCode, HttpHeaders responseHeaders) {

   390             throwing.accept(Where.BODY_HANDLER);

   391             BodySubscriber<T> subscriber = bodyHandler.apply(statusCode, responseHeaders);

   392             return new ThrowingBodySubscriber(throwing, subscriber);

   393         }

   394     }

   395 

   396     static class ThrowingBodySubscriber<T> implements BodySubscriber<T> {

   397         private final BodySubscriber<T> subscriber;

   398         volatile boolean onSubscribeCalled;

   399         final Consumer<Where> throwing;

   400         ThrowingBodySubscriber(Consumer<Where> throwing, BodySubscriber<T> subscriber) {

   401             this.throwing = throwing;

   402             this.subscriber = subscriber;

   403         }

   404 

   405         @Override

   406         public void onSubscribe(Flow.Subscription subscription) {

   407             //out.println("onSubscribe ");

   408             onSubscribeCalled = true;

   409             throwing.accept(Where.ON_SUBSCRIBE);

   410             subscriber.onSubscribe(subscription);

   411         }

   412 

   413         @Override

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

   415            // out.println("onNext " + item);

   416             assertTrue(onSubscribeCalled);

   417             throwing.accept(Where.ON_NEXT);

   418             subscriber.onNext(item);

   419         }

   420 

   421         @Override

   422         public void onError(Throwable throwable) {

   423             //out.println("onError");

   424             assertTrue(onSubscribeCalled);

   425             throwing.accept(Where.ON_ERROR);

   426             subscriber.onError(throwable);

   427         }

   428 

   429         @Override

   430         public void onComplete() {

   431             //out.println("onComplete");

   432             assertTrue(onSubscribeCalled, "onComplete called before onSubscribe");

   433             throwing.accept(Where.ON_COMPLETE);

   434             subscriber.onComplete();

   435         }

   436 

   437         @Override

   438         public CompletionStage<T> getBody() {

   439             throwing.accept(Where.GET_BODY);

   440             return subscriber.getBody();

   441         }

   442     }

   443 

   444 

   445     @BeforeTest

   446     public void setup() throws Exception {

   447         sslContext = new SimpleSSLContext().get();

   448         if (sslContext == null)

   449             throw new AssertionError("Unexpected null sslContext");

   450 

   451         // HTTP/1.1

   452         HttpTestHandler h1_fixedLengthHandler = new HTTP_FixedLengthHandler();

   453         HttpTestHandler h1_chunkHandler = new HTTP_ChunkedHandler();

   454         InetSocketAddress sa = new InetSocketAddress(0);

   455         httpTestServer = HttpTestServer.of(HttpServer.create(sa, 0));

   456         httpTestServer.addHandler(h1_fixedLengthHandler, "/http1/fixed");

   457         httpTestServer.addHandler(h1_chunkHandler, "/http1/chunk");

   458         httpURI_fixed = "http://127.0.0.1:" + httpTestServer.getAddress().getPort() + "/http1/fixed/x";

   459         httpURI_chunk = "http://127.0.0.1:" + httpTestServer.getAddress().getPort() + "/http1/chunk/x";

   460 

   461         HttpsServer httpsServer = HttpsServer.create(sa, 0);

   462         httpsServer.setHttpsConfigurator(new HttpsConfigurator(sslContext));

   463         httpsTestServer = HttpTestServer.of(httpsServer);

   464         httpsTestServer.addHandler(h1_fixedLengthHandler, "/https1/fixed");

   465         httpsTestServer.addHandler(h1_chunkHandler, "/https1/chunk");

   466         httpsURI_fixed = "https://127.0.0.1:" + httpsTestServer.getAddress().getPort() + "/https1/fixed/x";

   467         httpsURI_chunk = "https://127.0.0.1:" + httpsTestServer.getAddress().getPort() + "/https1/chunk/x";

   468 

   469         // HTTP/2

   470         HttpTestHandler h2_fixedLengthHandler = new HTTP_FixedLengthHandler();

   471         HttpTestHandler h2_chunkedHandler = new HTTP_ChunkedHandler();

   472 

   473         http2TestServer = HttpTestServer.of(new Http2TestServer("127.0.0.1", false, 0));

   474         http2TestServer.addHandler(h2_fixedLengthHandler, "/http2/fixed");

   475         http2TestServer.addHandler(h2_chunkedHandler, "/http2/chunk");

   476         int port = http2TestServer.getAddress().getPort();

   477         http2URI_fixed = "http://127.0.0.1:" + port + "/http2/fixed/x";

   478         http2URI_chunk = "http://127.0.0.1:" + port + "/http2/chunk/x";

   479 

   480         https2TestServer = HttpTestServer.of(new Http2TestServer("127.0.0.1", true, 0));

   481         https2TestServer.addHandler(h2_fixedLengthHandler, "/https2/fixed");

   482         https2TestServer.addHandler(h2_chunkedHandler, "/https2/chunk");

   483         port = https2TestServer.getAddress().getPort();

   484         https2URI_fixed = "https://127.0.0.1:" + port + "/https2/fixed/x";

   485         https2URI_chunk = "https://127.0.0.1:" + port + "/https2/chunk/x";

   486 

   487         httpTestServer.start();

   488         httpsTestServer.start();

   489         http2TestServer.start();

   490         https2TestServer.start();

   491     }

   492 

   493     @AfterTest

   494     public void teardown() throws Exception {

   495         httpTestServer.stop();

   496         httpsTestServer.stop();

   497         http2TestServer.stop();

   498         https2TestServer.stop();

   499     }

   500 

   501     static class HTTP_FixedLengthHandler implements HttpTestHandler {

   502         @Override

   503         public void handle(HttpTestExchange t) throws IOException {

   504             out.println("HTTP_FixedLengthHandler received request to " + t.getRequestURI());

   505             try (InputStream is = t.getRequestBody()) {

   506                 is.readAllBytes();

   507             }

   508             byte[] resp = t.getRequestURI().toString().getBytes(StandardCharsets.UTF_8);

   509             t.sendResponseHeaders(200, resp.length);  //fixed content length

   510             try (OutputStream os = t.getResponseBody()) {

   511                 os.write(resp);

   512             }

   513         }

   514     }

   515 

   516     static class HTTP_ChunkedHandler implements HttpTestHandler {

   517         @Override

   518         public void handle(HttpTestExchange t) throws IOException {

   519             out.println("HTTP_ChunkedHandler received request to " + t.getRequestURI());

   520             byte[] resp = t.getRequestURI().toString().getBytes(StandardCharsets.UTF_8);

   521             try (InputStream is = t.getRequestBody()) {

   522                 is.readAllBytes();

   523             }

   524             t.sendResponseHeaders(200, -1); // chunked/variable

   525             try (OutputStream os = t.getResponseBody()) {

   526                 os.write(resp);

   527             }

   528         }

   529     }

   530 

   531 }