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 import jdk.incubator.http.HttpResponse.BodySubscriber;

    25 import org.testng.annotations.DataProvider;

    26 import org.testng.annotations.Test;

    27 

    28 import java.io.BufferedReader;

    29 import java.io.ByteArrayInputStream;

    30 import java.io.InputStreamReader;

    31 import java.io.StringReader;

    32 import java.nio.ByteBuffer;

    33 import java.nio.charset.Charset;

    34 import java.nio.charset.MalformedInputException;

    35 import java.util.Arrays;

    36 import java.util.List;

    37 import java.util.concurrent.ExecutionException;

    38 import java.util.concurrent.SubmissionPublisher;

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

    40 import java.util.stream.Collectors;

    41 import java.util.stream.Stream;

    42 

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

    44 import static java.nio.charset.StandardCharsets.UTF_16;

    45 import static org.testng.Assert.assertEquals;

    46 import static org.testng.Assert.assertThrows;

    47 import static org.testng.Assert.assertTrue;

    48 

    49 /*

    50  * @test

    51  * @summary tests for BodySubscribers returned by asLines.

    52  *       In particular tests that surrogate characters are handled

    53  *       correctly.

    54  * @modules jdk.incubator.httpclient java.logging

    55  * @run testng/othervm LineStreamsAndSurrogatesTest

    56  */

    57 

    58 public class LineStreamsAndSurrogatesTest {

    59 

    60 

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

    62 

    63     private static final List<String> lines(String text) {

    64         return new BufferedReader(new StringReader(text)).lines().collect(Collectors.toList());

    65     }

    66 

    67     @Test

    68     void testUncomplete() throws Exception {

    69         // Uses U+10400 which is encoded as the surrogate pair U+D801 U+DC00

    70         String text = "Bient\u00f4t\r\n nous plongerons\r\n dans\r les\n\n" +

    71                 " fr\u00f4\ud801\udc00des\r\n t\u00e9n\u00e8bres\ud801\udc00";

    72         Charset charset = UTF_8;

    73 

    74         BodySubscriber<Stream<String>> bodySubscriber =

    75                 BodySubscriber.asLines(charset);

    76         AtomicReference<Throwable> errorRef = new AtomicReference<>();

    77         Runnable run = () -> {

    78             try {

    79                 SubmissionPublisher<List<ByteBuffer>> publisher = new SubmissionPublisher<>();

    80                 byte[] sbytes = text.getBytes(charset);

    81                 byte[] bytes = Arrays.copyOfRange(sbytes, 0, sbytes.length - 1);

    82                 publisher.subscribe(bodySubscriber);

    83                 System.out.println("Publishing " + bytes.length + " bytes");

    84                 for (int i = 0; i < bytes.length; i++) {

    85                     // ensure that surrogates are split over several buffers.

    86                     publisher.submit(List.of(ByteBuffer.wrap(bytes, i, 1)));

    87                 }

    88                 publisher.close();

    89             } catch(Throwable t) {

    90                 errorRef.set(t);

    91             }

    92         };

    93         Thread thread = new Thread(run,"Publishing");

    94         thread.start();

    95         try {

    96             Stream<String> stream = bodySubscriber.getBody().toCompletableFuture().get();

    97             List<String> list = stream.collect(Collectors.toList());

    98             String resp = list.stream().collect(Collectors.joining(""));

    99             System.out.println("***** Got: " + resp);

   100 

   101             byte[] sbytes = text.getBytes(UTF_8);

   102             byte[] bytes = Arrays.copyOfRange(sbytes, 0, sbytes.length - 1);

   103             ByteArrayInputStream bais = new ByteArrayInputStream(bytes);

   104             BufferedReader reader = new BufferedReader(new InputStreamReader(bais, charset));

   105             String resp2 = reader.lines().collect(Collectors.joining(""));

   106             System.out.println("***** Got2: " + resp2);

   107 

   108             assertEquals(resp, resp2);

   109             assertEquals(list, List.of("Bient\u00f4t",

   110                                        " nous plongerons",

   111                                        " dans",

   112                                        " les",

   113                                        "",

   114                                        " fr\u00f4\ud801\udc00des",

   115                                        " t\u00e9n\u00e8bres\ufffd"));

   116         } catch (ExecutionException x) {

   117             Throwable cause = x.getCause();

   118             if (cause instanceof MalformedInputException) {

   119                 throw new RuntimeException("Unexpected MalformedInputException", cause);

   120             }

   121             throw x;

   122         }

   123         if (errorRef.get() != null) {

   124             throw new RuntimeException("Unexpected exception", errorRef.get());

   125         }

   126     }

   127 

   128     @Test

   129     void testStream1() throws Exception {

   130         // Uses U+10400 which is encoded as the surrogate pair U+D801 U+DC00

   131         String text = "Bient\u00f4t\r\n nous plongerons\r\n dans\r\r les\n\n" +

   132                 " fr\u00f4\ud801\udc00des\r\n t\u00e9n\u00e8bres";

   133         Charset charset = UTF_8;

   134 

   135         BodySubscriber<Stream<String>> bodySubscriber =

   136                 BodySubscriber.asLines(charset);

   137         SubmissionPublisher<List<ByteBuffer>> publisher = new SubmissionPublisher<>();

   138         byte[] bytes = text.getBytes(charset);

   139         AtomicReference<Throwable> errorRef = new AtomicReference<>();

   140         Runnable run = () -> {

   141             try {

   142                 publisher.subscribe(bodySubscriber);

   143                 System.out.println("Publishing " + bytes.length + " bytes");

   144                 for (int i = 0; i < bytes.length; i++) {

   145                     // ensure that surrogates are split over several buffers.

   146                     publisher.submit(List.of(ByteBuffer.wrap(bytes, i, 1)));

   147                 }

   148                 publisher.close();

   149             } catch(Throwable t) {

   150                 errorRef.set(t);

   151             }

   152         };

   153 

   154         Stream<String> stream = bodySubscriber.getBody().toCompletableFuture().get();

   155         Thread thread = new Thread(run,"Publishing");

   156         thread.start();

   157         List<String> list = stream.collect(Collectors.toList());

   158         String resp = list.stream().collect(Collectors.joining("|"));

   159         System.out.println("***** Got: " + resp);

   160         assertEquals(resp, text.replace("\r\n", "|")

   161                                .replace("\n","|")

   162                                .replace("\r","|"));

   163         assertEquals(list, List.of("Bient\u00f4t",

   164                 " nous plongerons",

   165                 " dans",

   166                 "",

   167                 " les",

   168                 "",

   169                 " fr\u00f4\ud801\udc00des",

   170                 " t\u00e9n\u00e8bres"));

   171         assertEquals(list, lines(text));

   172         if (errorRef.get() != null) {

   173             throw new RuntimeException("Unexpected exception", errorRef.get());

   174         }

   175     }

   176 

   177 

   178     @Test

   179     void testStream2() throws Exception {

   180         String text = "Bient\u00f4t\r\n nous plongerons\r\n dans\r\r" +

   181                 " les fr\u00f4\ud801\udc00des\r\n t\u00e9n\u00e8bres\r\r";

   182         Charset charset = UTF_8;

   183 

   184         BodySubscriber<Stream<String>> bodySubscriber = BodySubscriber.asLines(charset);

   185         SubmissionPublisher<List<ByteBuffer>> publisher = new SubmissionPublisher<>();

   186         byte[] bytes = text.getBytes(charset);

   187         AtomicReference<Throwable> errorRef = new AtomicReference<>();

   188         Runnable run = () -> {

   189             try {

   190                 publisher.subscribe(bodySubscriber);

   191                 System.out.println("Publishing " + bytes.length + " bytes");

   192                 for (int i = 0; i < bytes.length; i++) {

   193                     // ensure that surrogates are split over several buffers.

   194                     publisher.submit(List.of(ByteBuffer.wrap(bytes, i, 1)));

   195                 }

   196                 publisher.close();

   197             } catch(Throwable t) {

   198                 errorRef.set(t);

   199             }

   200         };

   201 

   202         Stream<String> stream = bodySubscriber.getBody().toCompletableFuture().get();

   203         Thread thread = new Thread(run,"Publishing");

   204         thread.start();

   205         List<String> list = stream.collect(Collectors.toList());

   206         String resp = list.stream().collect(Collectors.joining(""));

   207         System.out.println("***** Got: " + resp);

   208         String expected = Stream.of(text.split("\r\n|\r|\n"))

   209                 .collect(Collectors.joining(""));

   210         assertEquals(resp, expected);

   211         assertEquals(list, List.of("Bient\u00f4t",

   212                 " nous plongerons",

   213                 " dans",

   214                 "",

   215                 " les fr\u00f4\ud801\udc00des",

   216                 " t\u00e9n\u00e8bres",

   217                 ""));

   218         assertEquals(list, lines(text));

   219         if (errorRef.get() != null) {

   220             throw new RuntimeException("Unexpected exception", errorRef.get());

   221         }

   222     }

   223 

   224     @Test

   225     void testStream3_UTF16() throws Exception {

   226         // Uses U+10400 which is encoded as the surrogate pair U+D801 U+DC00

   227         String text = "Bient\u00f4t\r\n nous plongerons\r\n dans\r\r" +

   228                 " les\n\n fr\u00f4\ud801\udc00des\r\n t\u00e9n\u00e8bres";

   229         Charset charset = UTF_16;

   230 

   231         BodySubscriber<Stream<String>> bodySubscriber =

   232                 BodySubscriber.asLines(charset);

   233         SubmissionPublisher<List<ByteBuffer>> publisher = new SubmissionPublisher<>();

   234         byte[] bytes = text.getBytes(charset);

   235         AtomicReference<Throwable> errorRef = new AtomicReference<>();

   236         Runnable run = () -> {

   237             try {

   238                 publisher.subscribe(bodySubscriber);

   239                 System.out.println("Publishing " + bytes.length + " bytes");

   240                 for (int i = 0; i < bytes.length; i++) {

   241                     // ensure that surrogates are split over several buffers.

   242                     publisher.submit(List.of(ByteBuffer.wrap(bytes, i, 1)));

   243                 }

   244                 publisher.close();

   245             } catch(Throwable t) {

   246                 errorRef.set(t);

   247             }

   248         };

   249 

   250         Stream<String> stream = bodySubscriber.getBody().toCompletableFuture().get();

   251         Thread thread = new Thread(run,"Publishing");

   252         thread.start();

   253         List<String> list = stream.collect(Collectors.toList());

   254         String resp = list.stream().collect(Collectors.joining(""));

   255         System.out.println("***** Got: " + resp);

   256         assertEquals(resp, text.replace("\n","").replace("\r",""));

   257         assertEquals(list, List.of("Bient\u00f4t",

   258                 " nous plongerons",

   259                 " dans",

   260                 "",

   261                 " les",

   262                 "",

   263                 " fr\u00f4\ud801\udc00des",

   264                 " t\u00e9n\u00e8bres"));

   265         assertEquals(list, lines(text));

   266         if (errorRef.get() != null) {

   267             throw new RuntimeException("Unexpected exception", errorRef.get());

   268         }

   269     }

   270 

   271 

   272     @Test

   273     void testStream4_UTF16() throws Exception {

   274         String text = "Bient\u00f4t\r\n nous plongerons\r\n dans\r\r" +

   275                 " les fr\u00f4\ud801\udc00des\r\n t\u00e9n\u00e8bres\r\r";

   276         Charset charset = UTF_16;

   277 

   278         BodySubscriber<Stream<String>> bodySubscriber =

   279                 BodySubscriber.asLines(charset);

   280         SubmissionPublisher<List<ByteBuffer>> publisher = new SubmissionPublisher<>();

   281         byte[] bytes = text.getBytes(charset);

   282         AtomicReference<Throwable> errorRef = new AtomicReference<>();

   283         Runnable run = () -> {

   284             try {

   285                 publisher.subscribe(bodySubscriber);

   286                 System.out.println("Publishing " + bytes.length + " bytes");

   287                 for (int i = 0; i < bytes.length; i++) {

   288                     // ensure that surrogates are split over several buffers.

   289                     publisher.submit(List.of(ByteBuffer.wrap(bytes, i, 1)));

   290                 }

   291                 publisher.close();

   292             } catch(Throwable t) {

   293                 errorRef.set(t);

   294             }

   295         };

   296 

   297         Stream<String> stream = bodySubscriber.getBody().toCompletableFuture().get();

   298         Thread thread = new Thread(run,"Publishing");

   299         thread.start();

   300         List<String> list = stream.collect(Collectors.toList());

   301         String resp = list.stream().collect(Collectors.joining(""));

   302         System.out.println("***** Got: " + resp);

   303         String expected = Stream.of(text.split("\r\n|\r|\n"))

   304                 .collect(Collectors.joining(""));

   305         assertEquals(resp, expected);

   306         assertEquals(list, List.of("Bient\u00f4t",

   307                 " nous plongerons",

   308                 " dans",

   309                 "",

   310                 " les fr\u00f4\ud801\udc00des",

   311                 " t\u00e9n\u00e8bres",

   312                 ""));

   313         assertEquals(list, lines(text));

   314         if (errorRef.get() != null) {

   315             throw new RuntimeException("Unexpected exception", errorRef.get());

   316         }

   317     }

   318 

   319 }