1 /*

     2  * Copyright (c) 2015, 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 package jdk.internal.net.http;

    25 

    26 import java.net.*;

    27 import java.io.*;

    28 import java.nio.channels.*;

    29 import java.nio.ByteBuffer;

    30 import java.util.concurrent.CountDownLatch;

    31 import java.util.concurrent.atomic.AtomicInteger;

    32 import java.net.http.HttpClient;

    33 import java.net.http.HttpRequest;

    34 import java.net.http.HttpResponse;

    35 import org.testng.annotations.Test;

    36 import jdk.internal.net.http.websocket.RawChannel;

    37 import static java.lang.System.out;

    38 import static java.nio.charset.StandardCharsets.US_ASCII;

    39 import static java.util.concurrent.TimeUnit.SECONDS;

    40 import static java.net.http.HttpResponse.BodyHandler.discard;

    41 

    42 /**

    43  * Whitebox test of selector mechanics. Currently only a simple test

    44  * setting one read and one write event is done. It checks that the

    45  * write event occurs first, followed by the read event and then no

    46  * further events occur despite the conditions actually still existing.

    47  */

    48 @Test

    49 public class SelectorTest {

    50 

    51     AtomicInteger counter = new AtomicInteger();

    52     volatile boolean error;

    53     static final CountDownLatch finishingGate = new CountDownLatch(1);

    54     static volatile HttpClient staticDefaultClient;

    55 

    56     static HttpClient defaultClient() {

    57         if (staticDefaultClient == null) {

    58             synchronized (SelectorTest.class) {

    59                 staticDefaultClient = HttpClient.newHttpClient();

    60             }

    61         }

    62         return staticDefaultClient;

    63     }

    64 

    65     String readSomeBytes(RawChannel chan) {

    66         try {

    67             ByteBuffer buf = chan.read();

    68             if (buf == null) {

    69                 out.println("chan read returned null");

    70                 return null;

    71             }

    72             buf.flip();

    73             byte[] bb = new byte[buf.remaining()];

    74             buf.get(bb);

    75             return new String(bb, US_ASCII);

    76         } catch (IOException ioe) {

    77             throw new UncheckedIOException(ioe);

    78         }

    79     }

    80 

    81     @Test

    82     public void test() throws Exception {

    83 

    84         try (ServerSocket server = new ServerSocket(0)) {

    85             int port = server.getLocalPort();

    86 

    87             out.println("Listening on port " + server.getLocalPort());

    88 

    89             TestServer t = new TestServer(server);

    90             t.start();

    91             out.println("Started server thread");

    92 

    93             try (RawChannel chan = getARawChannel(port)) {

    94 

    95                 chan.registerEvent(new RawChannel.RawEvent() {

    96                     @Override

    97                     public int interestOps() {

    98                         return SelectionKey.OP_READ;

    99                     }

   100 

   101                     @Override

   102                     public void handle() {

   103                         readSomeBytes(chan);

   104                         out.printf("OP_READ\n");

   105                         final int count = counter.get();

   106                         if (count != 1) {

   107                             out.printf("OP_READ error counter = %d\n", count);

   108                             error = true;

   109                         }

   110                     }

   111                 });

   112 

   113                 chan.registerEvent(new RawChannel.RawEvent() {

   114                     @Override

   115                     public int interestOps() {

   116                         return SelectionKey.OP_WRITE;

   117                     }

   118 

   119                     @Override

   120                     public void handle() {

   121                         out.printf("OP_WRITE\n");

   122                         final int count = counter.get();

   123                         if (count != 0) {

   124                             out.printf("OP_WRITE error counter = %d\n", count);

   125                             error = true;

   126                         } else {

   127                             ByteBuffer bb = ByteBuffer.wrap(TestServer.INPUT);

   128                             counter.incrementAndGet();

   129                             try {

   130                                 chan.write(new ByteBuffer[]{bb}, 0, 1);

   131                             } catch (IOException e) {

   132                                 throw new UncheckedIOException(e);

   133                             }

   134                         }

   135                     }

   136 

   137                 });

   138                 out.println("Events registered. Waiting");

   139                 finishingGate.await(30, SECONDS);

   140                 if (error)

   141                     throw new RuntimeException("Error");

   142                 else

   143                     out.println("No error");

   144             }

   145         }

   146     }

   147 

   148     static RawChannel getARawChannel(int port) throws Exception {

   149         URI uri = URI.create("http://127.0.0.1:" + port + "/");

   150         out.println("client connecting to " + uri.toString());

   151         HttpRequest req = HttpRequest.newBuilder(uri).build();

   152         // Otherwise HttpClient will think this is an ordinary connection and

   153         // thus all ordinary procedures apply to it, e.g. it must be put into

   154         // the cache

   155         ((HttpRequestImpl) req).isWebSocket(true);

   156         HttpResponse<?> r = defaultClient().send(req, discard());

   157         r.body();

   158         return ((HttpResponseImpl) r).rawChannel();

   159     }

   160 

   161     static class TestServer extends Thread {

   162         static final byte[] INPUT = "Hello world".getBytes(US_ASCII);

   163         static final byte[] OUTPUT = "Goodbye world".getBytes(US_ASCII);

   164         static final String FIRST_RESPONSE = "HTTP/1.1 200 OK\r\nContent-length: 0\r\n\r\n";

   165         final ServerSocket server;

   166 

   167         TestServer(ServerSocket server) throws IOException {

   168             this.server = server;

   169         }

   170 

   171         public void run() {

   172             try (Socket s = server.accept();

   173                  InputStream is = s.getInputStream();

   174                  OutputStream os = s.getOutputStream()) {

   175 

   176                 out.println("Got connection");

   177                 readRequest(is);

   178                 os.write(FIRST_RESPONSE.getBytes());

   179                 read(is);

   180                 write(os);

   181                 Thread.sleep(1000);

   182                 // send some more data, and make sure WRITE op does not get called

   183                 write(os);

   184                 out.println("TestServer exiting");

   185                 SelectorTest.finishingGate.countDown();

   186             } catch (Exception e) {

   187                 e.printStackTrace();

   188             }

   189         }

   190 

   191         // consumes the HTTP request

   192         static void readRequest(InputStream is) throws IOException {

   193             out.println("starting readRequest");

   194             byte[] buf = new byte[1024];

   195             String s = "";

   196             while (true) {

   197                 int n = is.read(buf);

   198                 if (n <= 0)

   199                     throw new IOException("Error");

   200                 s = s + new String(buf, 0, n);

   201                 if (s.indexOf("\r\n\r\n") != -1)

   202                     break;

   203             }

   204             out.println("returning from readRequest");

   205         }

   206 

   207         static void read(InputStream is) throws IOException {

   208             out.println("starting read");

   209             for (int i = 0; i < INPUT.length; i++) {

   210                 int c = is.read();

   211                 if (c == -1)

   212                     throw new IOException("closed");

   213                 if (INPUT[i] != (byte) c)

   214                     throw new IOException("Error. Expected:" + INPUT[i] + ", got:" + c);

   215             }

   216             out.println("returning from read");

   217         }

   218 

   219         static void write(OutputStream os) throws IOException {

   220             out.println("doing write");

   221             os.write(OUTPUT);

   222         }

   223     }

   224 }