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