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 java.io.ByteArrayOutputStream;

    25 import java.io.IOException;

    26 import java.io.InputStream;

    27 import java.io.OutputStream;

    28 import java.io.UncheckedIOException;

    29 import java.net.InetSocketAddress;

    30 import java.net.URI;

    31 import java.nio.ByteBuffer;

    32 import java.util.Collection;

    33 import java.util.List;

    34 import java.util.concurrent.Flow;

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

    36 import java.util.function.Function;

    37 import java.util.function.Supplier;

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

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

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

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

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

    43 import jdk.incubator.http.HttpClient;

    44 import jdk.incubator.http.HttpRequest;

    45 import jdk.incubator.http.HttpResponse;

    46 import jdk.incubator.http.HttpResponse.BodyHandler;

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

    48 import jdk.testlibrary.SimpleSSLContext;

    49 import org.testng.annotations.AfterTest;

    50 import org.testng.annotations.BeforeTest;

    51 import org.testng.annotations.DataProvider;

    52 import org.testng.annotations.Test;

    53 import javax.net.ssl.SSLContext;

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

    55 import static jdk.incubator.http.HttpRequest.BodyPublisher.fromString;

    56 import static org.testng.Assert.assertEquals;

    57 import static org.testng.Assert.assertThrows;

    58 import static org.testng.Assert.assertTrue;

    59 

    60 /*

    61  * @test

    62  * @summary Basic tests for Flow adapter Subscribers

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

    64  *          jdk.incubator.httpclient/jdk.incubator.http.internal.common

    65  *          jdk.incubator.httpclient/jdk.incubator.http.internal.frame

    66  *          jdk.incubator.httpclient/jdk.incubator.http.internal.hpack

    67  *          java.logging

    68  *          jdk.httpserver

    69  * @library /lib/testlibrary http2/server

    70  * @build Http2TestServer

    71  * @build jdk.testlibrary.SimpleSSLContext

    72  * @run testng/othervm FlowAdapterSubscriberTest

    73  */

    74 

    75 public class FlowAdapterSubscriberTest {

    76 

    77     SSLContext sslContext;

    78     HttpServer httpTestServer;         // HTTP/1.1    [ 4 servers ]

    79     HttpsServer httpsTestServer;       // HTTPS/1.1

    80     Http2TestServer http2TestServer;   // HTTP/2 ( h2c )

    81     Http2TestServer https2TestServer;  // HTTP/2 ( h2  )

    82     String httpURI;

    83     String httpsURI;

    84     String http2URI;

    85     String https2URI;

    86 

    87     @DataProvider(name = "uris")

    88     public Object[][] variants() {

    89         return new Object[][]{

    90                 { httpURI   },

    91                 { httpsURI  },

    92                 { http2URI  },

    93                 { https2URI },

    94         };

    95     }

    96 

    97     static final Class<NullPointerException> NPE = NullPointerException.class;

    98 

    99     @Test

   100     public void testNull() {

   101         assertThrows(NPE, () -> BodyHandler.fromSubscriber(null));

   102         assertThrows(NPE, () -> BodyHandler.fromSubscriber(null, Function.identity()));

   103         assertThrows(NPE, () -> BodyHandler.fromSubscriber(new ListSubscriber(), null));

   104         assertThrows(NPE, () -> BodyHandler.fromSubscriber(null, null));

   105 

   106         assertThrows(NPE, () -> BodySubscriber.fromSubscriber(null));

   107         assertThrows(NPE, () -> BodySubscriber.fromSubscriber(null, Function.identity()));

   108         assertThrows(NPE, () -> BodySubscriber.fromSubscriber(new ListSubscriber(), null));

   109         assertThrows(NPE, () -> BodySubscriber.fromSubscriber(null, null));

   110 

   111         Subscriber subscriber = BodySubscriber.fromSubscriber(new ListSubscriber());

   112         assertThrows(NPE, () -> subscriber.onSubscribe(null));

   113         assertThrows(NPE, () -> subscriber.onNext(null));

   114         assertThrows(NPE, () -> subscriber.onError(null));

   115     }

   116 

   117     // List<ByteBuffer>

   118 

   119     @Test(dataProvider = "uris")

   120     void testListWithFinisher(String url) {

   121         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   122         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   123                 .POST(fromString("May the luck of the Irish be with you!")).build();

   124 

   125         ListSubscriber subscriber = new ListSubscriber();

   126         HttpResponse<String> response = client.sendAsync(request,

   127                 BodyHandler.fromSubscriber(subscriber, Supplier::get)).join();

   128         String text = response.body();

   129         System.out.println(text);

   130         assertEquals(response.statusCode(), 200);

   131         assertEquals(text, "May the luck of the Irish be with you!");

   132     }

   133 

   134     @Test(dataProvider = "uris")

   135     void testListWithoutFinisher(String url) {

   136         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   137         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   138                 .POST(fromString("May the luck of the Irish be with you!")).build();

   139 

   140         ListSubscriber subscriber = new ListSubscriber();

   141         HttpResponse<Void> response = client.sendAsync(request,

   142                 BodyHandler.fromSubscriber(subscriber)).join();

   143         String text = subscriber.get();

   144         System.out.println(text);

   145         assertEquals(response.statusCode(), 200);

   146         assertEquals(text, "May the luck of the Irish be with you!");

   147     }

   148 

   149     @Test(dataProvider = "uris")

   150     void testListWithFinisherBlocking(String url) throws Exception {

   151         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   152         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   153                 .POST(fromString("May the luck of the Irish be with you!")).build();

   154 

   155         ListSubscriber subscriber = new ListSubscriber();

   156         HttpResponse<String> response = client.send(request,

   157                 BodyHandler.fromSubscriber(subscriber, Supplier::get));

   158         String text = response.body();

   159         System.out.println(text);

   160         assertEquals(response.statusCode(), 200);

   161         assertEquals(text, "May the luck of the Irish be with you!");

   162     }

   163 

   164     @Test(dataProvider = "uris")

   165     void testListWithoutFinisherBlocking(String url) throws Exception {

   166         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   167         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   168                 .POST(fromString("May the luck of the Irish be with you!")).build();

   169 

   170         ListSubscriber subscriber = new ListSubscriber();

   171         HttpResponse<Void> response = client.send(request,

   172                 BodyHandler.fromSubscriber(subscriber));

   173         String text = subscriber.get();

   174         System.out.println(text);

   175         assertEquals(response.statusCode(), 200);

   176         assertEquals(text, "May the luck of the Irish be with you!");

   177     }

   178 

   179     // Collection<ByteBuffer>

   180 

   181     @Test(dataProvider = "uris")

   182     void testCollectionWithFinisher(String url) {

   183         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   184         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   185                 .POST(fromString("What's the craic?")).build();

   186 

   187         CollectionSubscriber subscriber = new CollectionSubscriber();

   188         HttpResponse<String> response = client.sendAsync(request,

   189                 BodyHandler.fromSubscriber(subscriber, CollectionSubscriber::get)).join();

   190         String text = response.body();

   191         System.out.println(text);

   192         assertEquals(response.statusCode(), 200);

   193         assertEquals(text, "What's the craic?");

   194     }

   195 

   196     @Test(dataProvider = "uris")

   197     void testCollectionWithoutFinisher(String url) {

   198         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   199         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   200                 .POST(fromString("What's the craic?")).build();

   201 

   202         CollectionSubscriber subscriber = new CollectionSubscriber();

   203         HttpResponse<Void> response = client.sendAsync(request,

   204                 BodyHandler.fromSubscriber(subscriber)).join();

   205         String text = subscriber.get();

   206         System.out.println(text);

   207         assertEquals(response.statusCode(), 200);

   208         assertEquals(text, "What's the craic?");

   209     }

   210 

   211     @Test(dataProvider = "uris")

   212     void testCollectionWithFinisherBlocking(String url) throws Exception {

   213         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   214         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   215                 .POST(fromString("What's the craic?")).build();

   216 

   217         CollectionSubscriber subscriber = new CollectionSubscriber();

   218         HttpResponse<String> response = client.send(request,

   219                 BodyHandler.fromSubscriber(subscriber, CollectionSubscriber::get));

   220         String text = response.body();

   221         System.out.println(text);

   222         assertEquals(response.statusCode(), 200);

   223         assertEquals(text, "What's the craic?");

   224     }

   225 

   226     @Test(dataProvider = "uris")

   227     void testCollectionWithoutFinisheBlocking(String url) throws Exception {

   228         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   229         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   230                 .POST(fromString("What's the craic?")).build();

   231 

   232         CollectionSubscriber subscriber = new CollectionSubscriber();

   233         HttpResponse<Void> response = client.send(request,

   234                 BodyHandler.fromSubscriber(subscriber));

   235         String text = subscriber.get();

   236         System.out.println(text);

   237         assertEquals(response.statusCode(), 200);

   238         assertEquals(text, "What's the craic?");

   239     }

   240 

   241     // Iterable<ByteBuffer>

   242 

   243     @Test(dataProvider = "uris")

   244     void testIterableWithFinisher(String url) {

   245         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   246         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   247                 .POST(fromString("We're sucking diesel now!")).build();

   248 

   249         IterableSubscriber subscriber = new IterableSubscriber();

   250         HttpResponse<String> response = client.sendAsync(request,

   251                 BodyHandler.fromSubscriber(subscriber, Supplier::get)).join();

   252         String text = response.body();

   253         System.out.println(text);

   254         assertEquals(response.statusCode(), 200);

   255         assertEquals(text, "We're sucking diesel now!");

   256     }

   257 

   258     @Test(dataProvider = "uris")

   259     void testIterableWithoutFinisher(String url) {

   260         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   261         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   262                 .POST(fromString("We're sucking diesel now!")).build();

   263 

   264         IterableSubscriber subscriber = new IterableSubscriber();

   265         HttpResponse<Void> response = client.sendAsync(request,

   266                 BodyHandler.fromSubscriber(subscriber)).join();

   267         String text = subscriber.get();

   268         System.out.println(text);

   269         assertEquals(response.statusCode(), 200);

   270         assertEquals(text, "We're sucking diesel now!");

   271     }

   272 

   273     @Test(dataProvider = "uris")

   274     void testIterableWithFinisherBlocking(String url) throws Exception {

   275         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   276         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   277                 .POST(fromString("We're sucking diesel now!")).build();

   278 

   279         IterableSubscriber subscriber = new IterableSubscriber();

   280         HttpResponse<String> response = client.send(request,

   281                 BodyHandler.fromSubscriber(subscriber, Supplier::get));

   282         String text = response.body();

   283         System.out.println(text);

   284         assertEquals(response.statusCode(), 200);

   285         assertEquals(text, "We're sucking diesel now!");

   286     }

   287 

   288     @Test(dataProvider = "uris")

   289     void testIterableWithoutFinisherBlocking(String url) throws Exception{

   290         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   291         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   292                 .POST(fromString("We're sucking diesel now!")).build();

   293 

   294         IterableSubscriber subscriber = new IterableSubscriber();

   295         HttpResponse<Void> response = client.send(request,

   296                 BodyHandler.fromSubscriber(subscriber));

   297         String text = subscriber.get();

   298         System.out.println(text);

   299         assertEquals(response.statusCode(), 200);

   300         assertEquals(text, "We're sucking diesel now!");

   301     }

   302 

   303     // Subscriber<Object>

   304 

   305     @Test(dataProvider = "uris")

   306     void testObjectWithFinisher(String url) {

   307         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   308         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   309                 .POST(fromString("May the wind always be at your back.")).build();

   310 

   311         ObjectSubscriber subscriber = new ObjectSubscriber();

   312         HttpResponse<String> response = client.sendAsync(request,

   313                 BodyHandler.fromSubscriber(subscriber, ObjectSubscriber::get)).join();

   314         String text = response.body();

   315         System.out.println(text);

   316         assertEquals(response.statusCode(), 200);

   317         assertTrue(text.length() != 0);  // what else can be asserted!

   318     }

   319 

   320     @Test(dataProvider = "uris")

   321     void testObjectWithoutFinisher(String url) {

   322         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   323         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   324                 .POST(fromString("May the wind always be at your back.")).build();

   325 

   326         ObjectSubscriber subscriber = new ObjectSubscriber();

   327         HttpResponse<Void> response = client.sendAsync(request,

   328                 BodyHandler.fromSubscriber(subscriber)).join();

   329         String text = subscriber.get();

   330         System.out.println(text);

   331         assertEquals(response.statusCode(), 200);

   332         assertTrue(text.length() != 0);  // what else can be asserted!

   333     }

   334 

   335     @Test(dataProvider = "uris")

   336     void testObjectWithFinisherBlocking(String url) throws Exception {

   337         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   338         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   339                 .POST(fromString("May the wind always be at your back.")).build();

   340 

   341         ObjectSubscriber subscriber = new ObjectSubscriber();

   342         HttpResponse<String> response = client.send(request,

   343                 BodyHandler.fromSubscriber(subscriber, ObjectSubscriber::get));

   344         String text = response.body();

   345         System.out.println(text);

   346         assertEquals(response.statusCode(), 200);

   347         assertTrue(text.length() != 0);  // what else can be asserted!

   348     }

   349 

   350     @Test(dataProvider = "uris")

   351     void testObjectWithoutFinisherBlocking(String url) throws Exception {

   352         HttpClient client = HttpClient.newBuilder().sslContext(sslContext).build();

   353         HttpRequest request = HttpRequest.newBuilder(URI.create(url))

   354                 .POST(fromString("May the wind always be at your back.")).build();

   355 

   356         ObjectSubscriber subscriber = new ObjectSubscriber();

   357         HttpResponse<Void> response = client.send(request,

   358                 BodyHandler.fromSubscriber(subscriber));

   359         String text = subscriber.get();

   360         System.out.println(text);

   361         assertEquals(response.statusCode(), 200);

   362         assertTrue(text.length() != 0);  // what else can be asserted!

   363     }

   364 

   365     /** An abstract Subscriber that converts all received data into a String. */

   366     static abstract class AbstractSubscriber implements Supplier<String> {

   367         protected volatile Flow.Subscription subscription;

   368         protected volatile ByteArrayOutputStream baos = new ByteArrayOutputStream();

   369         protected volatile String text;

   370 

   371         public void onSubscribe(Flow.Subscription subscription) {

   372             this.subscription = subscription;

   373             subscription.request(Long.MAX_VALUE);

   374         }

   375         public void onError(Throwable throwable) {

   376             throw new RuntimeException(throwable);

   377         }

   378         public void onComplete() {

   379             text = new String(baos.toByteArray(), UTF_8);

   380         }

   381         @Override public String get() { return text; }

   382     }

   383 

   384     static class ListSubscriber extends AbstractSubscriber

   385         implements Flow.Subscriber<List<ByteBuffer>>, Supplier<String>

   386     {

   387         @Override public void onNext(List<ByteBuffer> item) {

   388             for (ByteBuffer bb : item) {

   389                 byte[] ba = new byte[bb.remaining()];

   390                 bb.get(ba);

   391                 uncheckedWrite(baos, ba);

   392             }

   393         }

   394     }

   395 

   396     static class CollectionSubscriber extends AbstractSubscriber

   397         implements Flow.Subscriber<Collection<ByteBuffer>>, Supplier<String>

   398     {

   399         @Override public void onNext(Collection<ByteBuffer> item) {

   400             for (ByteBuffer bb : item) {

   401                 byte[] ba = new byte[bb.remaining()];

   402                 bb.get(ba);

   403                 uncheckedWrite(baos, ba);

   404             }

   405         }

   406     }

   407 

   408     static class IterableSubscriber extends AbstractSubscriber

   409         implements Flow.Subscriber<Iterable<ByteBuffer>>, Supplier<String>

   410     {

   411         @Override public void onNext(Iterable<ByteBuffer> item) {

   412             for (ByteBuffer bb : item) {

   413                 byte[] ba = new byte[bb.remaining()];

   414                 bb.get(ba);

   415                 uncheckedWrite(baos, ba);

   416             }

   417         }

   418     }

   419 

   420     static class ObjectSubscriber extends AbstractSubscriber

   421         implements Flow.Subscriber<Object>, Supplier<String>

   422     {

   423         @Override public void onNext(Object item) {

   424             // What can anyone do with Object, cast or toString it ?

   425             uncheckedWrite(baos, item.toString().getBytes(UTF_8));

   426         }

   427     }

   428 

   429     static void uncheckedWrite(ByteArrayOutputStream baos, byte[] ba) {

   430         try {

   431             baos.write(ba);

   432         } catch (IOException e) {

   433             throw new UncheckedIOException(e);

   434         }

   435     }

   436 

   437     @BeforeTest

   438     public void setup() throws Exception {

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

   440         if (sslContext == null)

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

   442 

   443         InetSocketAddress sa = new InetSocketAddress("localhost", 0);

   444         httpTestServer = HttpServer.create(sa, 0);

   445         httpTestServer.createContext("/http1/echo", new Http1EchoHandler());

   446         httpURI = "http://127.0.0.1:" + httpTestServer.getAddress().getPort() + "/http1/echo";

   447 

   448         httpsTestServer = HttpsServer.create(sa, 0);

   449         httpsTestServer.setHttpsConfigurator(new HttpsConfigurator(sslContext));

   450         httpsTestServer.createContext("/https1/echo", new Http1EchoHandler());

   451         httpsURI = "https://127.0.0.1:" + httpsTestServer.getAddress().getPort() + "/https1/echo";

   452 

   453         http2TestServer = new Http2TestServer("127.0.0.1", false, 0);

   454         http2TestServer.addHandler(new Http2EchoHandler(), "/http2/echo");

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

   456         http2URI = "http://127.0.0.1:" + port + "/http2/echo";

   457 

   458         https2TestServer = new Http2TestServer("127.0.0.1", true, 0);

   459         https2TestServer.addHandler(new Http2EchoHandler(), "/https2/echo");

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

   461         https2URI = "https://127.0.0.1:" + port + "/https2/echo";

   462 

   463         httpTestServer.start();

   464         httpsTestServer.start();

   465         http2TestServer.start();

   466         https2TestServer.start();

   467     }

   468 

   469     @AfterTest

   470     public void teardown() throws Exception {

   471         httpTestServer.stop(0);

   472         httpsTestServer.stop(0);

   473         http2TestServer.stop();

   474         https2TestServer.stop();

   475     }

   476 

   477     static class Http1EchoHandler implements HttpHandler {

   478         @Override

   479         public void handle(HttpExchange t) throws IOException {

   480             try (InputStream is = t.getRequestBody();

   481                  OutputStream os = t.getResponseBody()) {

   482                 byte[] bytes = is.readAllBytes();

   483                 t.sendResponseHeaders(200, bytes.length);

   484                 os.write(bytes);

   485             }

   486         }

   487     }

   488 

   489     static class Http2EchoHandler implements Http2Handler {

   490         @Override

   491         public void handle(Http2TestExchange t) throws IOException {

   492             try (InputStream is = t.getRequestBody();

   493                  OutputStream os = t.getResponseBody()) {

   494                 byte[] bytes = is.readAllBytes();

   495                 t.sendResponseHeaders(200, bytes.length);

   496                 os.write(bytes);

   497             }

   498         }

   499     }

   500 }