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

    31 import java.io.IOException;

    32 import java.io.InputStreamReader;

    33 import java.io.StringReader;

    34 import java.io.UncheckedIOException;

    35 import java.nio.ByteBuffer;

    36 import java.nio.charset.MalformedInputException;

    37 import java.nio.charset.StandardCharsets;

    38 import java.util.ArrayList;

    39 import java.util.Arrays;

    40 import java.util.List;

    41 import java.util.concurrent.CopyOnWriteArrayList;

    42 import java.util.concurrent.ExecutionException;

    43 import java.util.concurrent.Flow;

    44 import java.util.concurrent.SubmissionPublisher;

    45 import java.util.function.Supplier;

    46 import java.util.stream.Collectors;

    47 import java.util.stream.Stream;

    48 

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

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

    51 import static org.testng.Assert.assertEquals;

    52 import static org.testng.Assert.assertThrows;

    53 import static org.testng.Assert.assertTrue;

    54 

    55 /*

    56  * @test

    57  * @summary tests for BodySubscribers returned by fromLineSubscriber.

    58  *       In particular tests that surrogate characters are handled

    59  *       correctly.

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

    61  * @run testng/othervm LineSubscribersAndSurrogatesTest

    62  */

    63 

    64 public class LineSubscribersAndSurrogatesTest {

    65 

    66 

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

    68 

    69     private static final List<String> lines(String text, String eol) {

    70         if (eol == null) {

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

    72         } else {

    73             String replaced = text.replace(eol, "|");

    74             int i=0;

    75             while(replaced.endsWith("||")) {

    76                 replaced = replaced.substring(0,replaced.length()-1);

    77                 i++;

    78             }

    79             List<String> res = List.of(replaced.split("\\|"));

    80             if (i > 0) {

    81                 res = new ArrayList<>(res);

    82                 for (int j=0; j<i; j++) res.add("");

    83             }

    84             return res;

    85         }

    86     }

    87 

    88     @Test

    89     void testIncomplete() throws Exception {

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

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

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

    93         ObjectSubscriber subscriber = new ObjectSubscriber();

    94         BodySubscriber<String> bodySubscriber = BodySubscriber.fromLineSubscriber(

    95                 subscriber, Supplier::get, UTF_8, null);

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

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

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

    99         publisher.subscribe(bodySubscriber);

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

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

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

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

   104         }

   105         publisher.close();

   106         try {

   107             String resp = bodySubscriber.getBody().toCompletableFuture().get();

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

   109             ByteArrayInputStream bais = new ByteArrayInputStream(bytes);

   110             BufferedReader reader = new BufferedReader(new InputStreamReader(bais, UTF_8));

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

   112             assertEquals(resp, resp2);

   113             assertEquals(subscriber.list, List.of("Bient\u00f4t",

   114                     " nous plongerons",

   115                     " dans",

   116                     " les",

   117                     "",

   118                     " fr\u00f4\ud801\udc00des",

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

   120         } catch (ExecutionException x) {

   121             Throwable cause = x.getCause();

   122             if (cause instanceof MalformedInputException) {

   123                 throw new RuntimeException("Unexpected MalformedInputException thrown", cause);

   124             }

   125             throw x;

   126         }

   127     }

   128 

   129 

   130     @Test

   131     void testStringWithFinisherLF() throws Exception {

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

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

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

   135         ObjectSubscriber subscriber = new ObjectSubscriber();

   136         BodySubscriber<String> bodySubscriber = BodySubscriber.fromLineSubscriber(

   137                 subscriber, Supplier::get, UTF_8, "\n");

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

   139         byte[] bytes = text.getBytes(UTF_8);

   140         publisher.subscribe(bodySubscriber);

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

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

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

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

   145         }

   146         publisher.close();

   147         String resp = bodySubscriber.getBody().toCompletableFuture().get();

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

   149         List<String> expected = List.of("Bient\u00f4t\r",

   150                 " nous plongerons\r",

   151                 " dans\r les",

   152                 "",

   153                 " fr\u00f4\ud801\udc00des\r",

   154                 " t\u00e9n\u00e8bres\r");

   155         assertEquals(subscriber.list, expected);

   156         assertEquals(resp, Stream.of(text.split("\n")).collect(Collectors.joining("")));

   157         assertEquals(resp, expected.stream().collect(Collectors.joining("")));

   158         assertEquals(subscriber.list, lines(text, "\n"));

   159     }

   160 

   161 

   162     @Test

   163     void testStringWithFinisherCR() throws Exception {

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

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

   166         ObjectSubscriber subscriber = new ObjectSubscriber();

   167         BodySubscriber<String> bodySubscriber = BodySubscriber.fromLineSubscriber(

   168                 subscriber, Supplier::get, UTF_8, "\r");

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

   170         byte[] bytes = text.getBytes(UTF_8);

   171         publisher.subscribe(bodySubscriber);

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

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

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

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

   176         }

   177         publisher.close();

   178         String resp = bodySubscriber.getBody().toCompletableFuture().get();

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

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

   181         assertEquals(subscriber.list, List.of("Bient\u00f4t",

   182                 "\n nous plongerons",

   183                 "\n dans",

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

   185                 "\n t\u00e9n\u00e8bres",

   186                 ""));

   187         assertEquals(subscriber.list, lines(text, "\r"));

   188     }

   189 

   190     @Test

   191     void testStringWithFinisherCRLF() throws Exception {

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

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

   194         ObjectSubscriber subscriber = new ObjectSubscriber();

   195         BodySubscriber<String> bodySubscriber = BodySubscriber.fromLineSubscriber(

   196                 subscriber, Supplier::get, UTF_8, "\r\n");

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

   198         byte[] bytes = text.getBytes(UTF_8);

   199         publisher.subscribe(bodySubscriber);

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

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

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

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

   204         }

   205         publisher.close();

   206         String resp = bodySubscriber.getBody().toCompletableFuture().get();

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

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

   209         assertEquals(subscriber.list, List.of("Bient\u00f4t",

   210                 " nous plongerons",

   211                 " dans\r les fr\u00f4\ud801\udc00des",

   212                 " t\u00e9n\u00e8bres"));

   213         assertEquals(subscriber.list, lines(text, "\r\n"));

   214     }

   215 

   216 

   217     @Test

   218     void testStringWithFinisherBR() throws Exception {

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

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

   221         ObjectSubscriber subscriber = new ObjectSubscriber();

   222         BodySubscriber<String> bodySubscriber = BodySubscriber.fromLineSubscriber(

   223                 subscriber, Supplier::get, UTF_8, null);

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

   225         byte[] bytes = text.getBytes(UTF_8);

   226         publisher.subscribe(bodySubscriber);

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

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

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

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

   231         }

   232         publisher.close();

   233         String resp = bodySubscriber.getBody().toCompletableFuture().get();

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

   235         List<String> expected = List.of("Bient\u00f4t",

   236                 " nous plongerons",

   237                 " dans",

   238                 " les",

   239                 "",

   240                 " fr\u00f4\ud801\udc00des",

   241                 " t\u00e9n\u00e8bres");

   242         assertEquals(subscriber.list, expected);

   243         assertEquals(resp, expected.stream().collect(Collectors.joining("")));

   244         assertEquals(subscriber.list, lines(text, null));

   245     }

   246 

   247     @Test

   248     void testStringWithFinisherBR_UTF_16() throws Exception {

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

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

   251         ObjectSubscriber subscriber = new ObjectSubscriber();

   252         BodySubscriber<String> bodySubscriber = BodySubscriber.fromLineSubscriber(

   253                 subscriber, Supplier::get, UTF_16, null);

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

   255         byte[] bytes = text.getBytes(UTF_16);

   256         publisher.subscribe(bodySubscriber);

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

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

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

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

   261         }

   262         publisher.close();

   263         String resp = bodySubscriber.getBody().toCompletableFuture().get();

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

   265         List<String> expected = List.of("Bient\u00f4t",

   266                 " nous plongerons",

   267                 " dans",

   268                 " les",

   269                 "",

   270                 " fr\u00f4\ud801\udc00des",

   271                 " t\u00e9n\u00e8bres",

   272                 "");

   273         assertEquals(resp, expected.stream().collect(Collectors.joining("")));

   274         assertEquals(subscriber.list, expected);

   275         assertEquals(subscriber.list, lines(text, null));

   276     }

   277 

   278     void testStringWithoutFinisherBR() throws Exception {

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

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

   281         ObjectSubscriber subscriber = new ObjectSubscriber();

   282         BodySubscriber<Void> bodySubscriber = BodySubscriber.fromLineSubscriber(subscriber);

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

   284         byte[] bytes = text.getBytes(UTF_8);

   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         Void resp = bodySubscriber.getBody().toCompletableFuture().get();

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

   294         List<String> expected = List.of("Bient\u00f4t",

   295                 " nous plongerons",

   296                 " dans",

   297                 " les",

   298                 "",

   299                 " fr\u00f4\ud801\udc00des",

   300                 " t\u00e9n\u00e8bres");

   301         assertEquals(subscriber.text, expected.stream().collect(Collectors.joining("")));

   302         assertEquals(subscriber.list, expected);

   303         assertEquals(subscriber.list, lines(text, null));

   304     }

   305 

   306 

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

   308     static abstract class AbstractSubscriber implements Supplier<String> {

   309         protected final List<Object> list = new CopyOnWriteArrayList<>();

   310         protected volatile Flow.Subscription subscription;

   311         protected final StringBuilder baos = new StringBuilder();

   312         protected volatile String text;

   313         protected volatile RuntimeException error;

   314 

   315         public void onSubscribe(Flow.Subscription subscription) {

   316             this.subscription = subscription;

   317             subscription.request(Long.MAX_VALUE);

   318         }

   319         public void onError(Throwable throwable) {

   320             System.out.println(this + " onError: " + throwable);

   321             error = new RuntimeException(throwable);

   322         }

   323         public void onComplete() {

   324             System.out.println(this + " onComplete");

   325             text = baos.toString();

   326         }

   327         @Override public String get() {

   328             if (error != null) throw error;

   329             return text;

   330         }

   331         public final List<?> list() {

   332             return list;

   333         }

   334     }

   335 

   336     static class StringSubscriber extends AbstractSubscriber

   337             implements Flow.Subscriber<String>, Supplier<String>

   338     {

   339         @Override public void onNext(String item) {

   340             System.out.println(this + " onNext: \""

   341                     + item.replace("\n","\\n")

   342                           .replace("\r", "\\r")

   343                     + "\"");

   344             baos.append(item);

   345             list.add(item);

   346         }

   347     }

   348 

   349     static class CharSequenceSubscriber extends AbstractSubscriber

   350             implements Flow.Subscriber<CharSequence>, Supplier<String>

   351     {

   352         @Override public void onNext(CharSequence item) {

   353             System.out.println(this + " onNext: \""

   354                     + item.toString().replace("\n","\\n")

   355                     .replace("\r", "\\r")

   356                     + "\"");

   357             baos.append(item);

   358             list.add(item);

   359         }

   360     }

   361 

   362     static class ObjectSubscriber extends AbstractSubscriber

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

   364     {

   365         @Override public void onNext(Object item) {

   366             System.out.println(this + " onNext: \""

   367                     + item.toString().replace("\n","\\n")

   368                     .replace("\r", "\\r")

   369                     + "\"");

   370             baos.append(item);

   371             list.add(item);

   372         }

   373     }

   374 

   375 

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

   377         try {

   378             baos.write(ba);

   379         } catch (IOException e) {

   380             throw new UncheckedIOException(e);

   381         }

   382     }

   383 

   384 }