1 /*
   2  * Copyright (c) 2015, 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 package jdk.incubator.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 static java.lang.System.out;
  33 import static java.nio.charset.StandardCharsets.US_ASCII;
  34 import static java.util.concurrent.TimeUnit.SECONDS;
  35 import static jdk.incubator.http.HttpResponse.BodyHandler.discard;
  36 
  37 import org.testng.annotations.Test;
  38 import jdk.incubator.http.internal.websocket.RawChannel;
  39 
  40 /**
  41  * Whitebox test of selector mechanics. Currently only a simple test
  42  * setting one read and one write event is done. It checks that the
  43  * write event occurs first, followed by the read event and then no
  44  * further events occur despite the conditions actually still existing.
  45  */
  46 @Test
  47 public class SelectorTest {
  48 
  49     AtomicInteger counter = new AtomicInteger();
  50     volatile boolean error;
  51     static final CountDownLatch finishingGate = new CountDownLatch(1);
  52     static volatile HttpClient staticDefaultClient;
  53 
  54     static HttpClient defaultClient() {
  55         if (staticDefaultClient == null) {
  56             synchronized (SelectorTest.class) {
  57                 staticDefaultClient = HttpClient.newHttpClient();
  58             }
  59         }
  60         return staticDefaultClient;
  61     }
  62 
  63     String readSomeBytes(RawChannel chan) {
  64         try {
  65             ByteBuffer buf = chan.read();
  66             if (buf == null) {
  67                 out.println("chan read returned null");
  68                 return null;
  69             }
  70             buf.flip();
  71             byte[] bb = new byte[buf.remaining()];
  72             buf.get(bb);
  73             return new String(bb, US_ASCII);
  74         } catch (IOException ioe) {
  75             throw new UncheckedIOException(ioe);
  76         }
  77     }
  78 
  79     @Test
  80     public void test() throws Exception {
  81 
  82         try (ServerSocket server = new ServerSocket(0)) {
  83             int port = server.getLocalPort();
  84 
  85             out.println("Listening on port " + server.getLocalPort());
  86 
  87             TestServer t = new TestServer(server);
  88             t.start();
  89             out.println("Started server thread");
  90 
  91             final RawChannel chan = getARawChannel(port);
  92 
  93             chan.registerEvent(new RawChannel.RawEvent() {
  94                 @Override
  95                 public int interestOps() {
  96                     return SelectionKey.OP_READ;
  97                 }
  98 
  99                 @Override
 100                 public void handle() {
 101                     readSomeBytes(chan);
 102                     out.printf("OP_READ\n");
 103                     final int count = counter.get();
 104                     if (count != 1) {
 105                         out.printf("OP_READ error counter = %d\n", count);
 106                         error = true;
 107                     }
 108                 }
 109             });
 110 
 111             chan.registerEvent(new RawChannel.RawEvent() {
 112                 @Override
 113                 public int interestOps() {
 114                     return SelectionKey.OP_WRITE;
 115                 }
 116 
 117                 @Override
 118                 public void handle() {
 119                     out.printf("OP_WRITE\n");
 120                     final int count = counter.get();
 121                     if (count != 0) {
 122                         out.printf("OP_WRITE error counter = %d\n", count);
 123                         error = true;
 124                     } else {
 125                         ByteBuffer bb = ByteBuffer.wrap(TestServer.INPUT);
 126                         counter.incrementAndGet();
 127                         try {
 128                             chan.write(new ByteBuffer[]{bb}, 0, 1);
 129                         } catch (IOException e) {
 130                             throw new UncheckedIOException(e);
 131                         }
 132                     }
 133                 }
 134 
 135             });
 136             out.println("Events registered. Waiting");
 137             finishingGate.await(30, SECONDS);
 138             if (error)
 139                 throw new RuntimeException("Error");
 140             else
 141                 out.println("No error");
 142         }
 143     }
 144 
 145     static RawChannel getARawChannel(int port) throws Exception {
 146         URI uri = URI.create("http://127.0.0.1:" + port + "/");
 147         out.println("client connecting to " + uri.toString());
 148         HttpRequest req = HttpRequest.newBuilder(uri).build();
 149         HttpResponse<?> r = defaultClient().send(req, discard(null));
 150         r.body();
 151         return ((HttpResponseImpl) r).rawChannel();
 152     }
 153 
 154     static class TestServer extends Thread {
 155         static final byte[] INPUT = "Hello world".getBytes(US_ASCII);
 156         static final byte[] OUTPUT = "Goodbye world".getBytes(US_ASCII);
 157         static final String FIRST_RESPONSE = "HTTP/1.1 200 OK\r\nContent-length: 0\r\n\r\n";
 158         final ServerSocket server;
 159 
 160         TestServer(ServerSocket server) throws IOException {
 161             this.server = server;
 162         }
 163 
 164         public void run() {
 165             try (Socket s = server.accept();
 166                  InputStream is = s.getInputStream();
 167                  OutputStream os = s.getOutputStream()) {
 168 
 169                 out.println("Got connection");
 170                 readRequest(is);
 171                 os.write(FIRST_RESPONSE.getBytes());
 172                 read(is);
 173                 write(os);
 174                 Thread.sleep(1000);
 175                 // send some more data, and make sure WRITE op does not get called
 176                 write(os);
 177                 out.println("TestServer exiting");
 178                 SelectorTest.finishingGate.countDown();
 179             } catch (Exception e) {
 180                 e.printStackTrace();
 181             }
 182         }
 183 
 184         // consumes the HTTP request
 185         static void readRequest(InputStream is) throws IOException {
 186             out.println("starting readRequest");
 187             byte[] buf = new byte[1024];
 188             String s = "";
 189             while (true) {
 190                 int n = is.read(buf);
 191                 if (n <= 0)
 192                     throw new IOException("Error");
 193                 s = s + new String(buf, 0, n);
 194                 if (s.indexOf("\r\n\r\n") != -1)
 195                     break;
 196             }
 197             out.println("returning from readRequest");
 198         }
 199 
 200         static void read(InputStream is) throws IOException {
 201             out.println("starting read");
 202             for (int i = 0; i < INPUT.length; i++) {
 203                 int c = is.read();
 204                 if (c == -1)
 205                     throw new IOException("closed");
 206                 if (INPUT[i] != (byte) c)
 207                     throw new IOException("Error. Expected:" + INPUT[i] + ", got:" + c);
 208             }
 209             out.println("returning from read");
 210         }
 211 
 212         static void write(OutputStream os) throws IOException {
 213             out.println("doing write");
 214             os.write(OUTPUT);
 215         }
 216     }
 217 }