1 /*
   2  * Copyright (c) 2008, 2011, 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 /* @test
  25  * @bug 4607272 6842687 6878369 6944810 7023403
  26  * @summary Unit test for AsynchronousSocketChannel
  27  * @run main Basic -skipSlowConnectTest

  28  */
  29 
  30 import java.nio.ByteBuffer;
  31 import java.nio.channels.*;
  32 import static java.net.StandardSocketOptions.*;
  33 import java.net.*;
  34 import java.util.Random;
  35 import java.util.concurrent.*;
  36 import java.util.concurrent.atomic.*;
  37 import java.io.Closeable;
  38 import java.io.IOException;
  39 
  40 public class Basic {
  41     static final Random rand = new Random();
  42 
  43     static boolean skipSlowConnectTest = false;
  44 
  45     public static void main(String[] args) throws Exception {
  46         for (String arg: args) {
  47             switch (arg) {
  48             case "-skipSlowConnectTest" :
  49                 skipSlowConnectTest = true;
  50                 break;
  51             default:
  52                 throw new RuntimeException("Unrecognized argument: " + arg);
  53             }
  54         }
  55 
  56         testBind();
  57         testSocketOptions();
  58         testConnect();
  59         testCloseWhenPending();
  60         testCancel();
  61         testRead1();
  62         testRead2();
  63         testRead3();
  64         testWrite1();
  65         testWrite2();
  66         // skip timeout tests until 7052549 is fixed
  67         if (!System.getProperty("os.name").startsWith("Windows"))
  68             testTimeout();
  69         testShutdown();
  70     }
  71 
  72     static class Server implements Closeable {
  73         private final ServerSocketChannel ssc;
  74         private final InetSocketAddress address;
  75 
  76         Server() throws IOException {
  77             ssc = ServerSocketChannel.open().bind(new InetSocketAddress(0));
  78 
  79             InetAddress lh = InetAddress.getLocalHost();
  80             int port = ((InetSocketAddress)(ssc.getLocalAddress())).getPort();
  81             address = new InetSocketAddress(lh, port);
  82         }
  83 
  84         InetSocketAddress address() {
  85             return address;
  86         }
  87 
  88         SocketChannel accept() throws IOException {
  89             return ssc.accept();
  90         }
  91 
  92         public void close() throws IOException {
  93             ssc.close();
  94         }
  95 
  96     }
  97 
  98     static void testBind() throws Exception {
  99         System.out.println("-- bind --");
 100 
 101         try (AsynchronousSocketChannel ch = AsynchronousSocketChannel.open()) {
 102             if (ch.getLocalAddress() != null)
 103                 throw new RuntimeException("Local address should be 'null'");
 104             ch.bind(new InetSocketAddress(0));
 105 
 106             // check local address after binding
 107             InetSocketAddress local = (InetSocketAddress)ch.getLocalAddress();
 108             if (local.getPort() == 0)
 109                 throw new RuntimeException("Unexpected port");
 110             if (!local.getAddress().isAnyLocalAddress())
 111                 throw new RuntimeException("Not bound to a wildcard address");
 112 
 113             // try to re-bind
 114             try {
 115                 ch.bind(new InetSocketAddress(0));
 116                 throw new RuntimeException("AlreadyBoundException expected");
 117             } catch (AlreadyBoundException x) {
 118             }
 119         }
 120 
 121         // check ClosedChannelException
 122         AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 123         ch.close();
 124         try {
 125             ch.bind(new InetSocketAddress(0));
 126             throw new RuntimeException("ClosedChannelException  expected");
 127         } catch (ClosedChannelException  x) {
 128         }
 129     }
 130 
 131     static void testSocketOptions() throws Exception {
 132         System.out.println("-- socket options --");
 133 
 134         try (AsynchronousSocketChannel ch = AsynchronousSocketChannel.open()) {
 135             ch.setOption(SO_RCVBUF, 128*1024)
 136               .setOption(SO_SNDBUF, 128*1024)
 137               .setOption(SO_REUSEADDR, true);
 138 
 139             // check SO_SNDBUF/SO_RCVBUF limits
 140             int before, after;
 141             before = ch.getOption(SO_SNDBUF);
 142             after = ch.setOption(SO_SNDBUF, Integer.MAX_VALUE).getOption(SO_SNDBUF);
 143             if (after < before)
 144                 throw new RuntimeException("setOption caused SO_SNDBUF to decrease");
 145             before = ch.getOption(SO_RCVBUF);
 146             after = ch.setOption(SO_RCVBUF, Integer.MAX_VALUE).getOption(SO_RCVBUF);
 147             if (after < before)
 148                 throw new RuntimeException("setOption caused SO_RCVBUF to decrease");
 149 
 150             ch.bind(new InetSocketAddress(0));
 151 
 152             // default values
 153             if (ch.getOption(SO_KEEPALIVE))
 154                 throw new RuntimeException("Default of SO_KEEPALIVE should be 'false'");
 155             if (ch.getOption(TCP_NODELAY))
 156                 throw new RuntimeException("Default of TCP_NODELAY should be 'false'");
 157 
 158             // set and check
 159             if (!ch.setOption(SO_KEEPALIVE, true).getOption(SO_KEEPALIVE))
 160                 throw new RuntimeException("SO_KEEPALIVE did not change");
 161             if (!ch.setOption(TCP_NODELAY, true).getOption(TCP_NODELAY))
 162                 throw new RuntimeException("SO_KEEPALIVE did not change");
 163 
 164             // read others (can't check as actual value is implementation dependent)
 165             ch.getOption(SO_RCVBUF);
 166             ch.getOption(SO_SNDBUF);
 167         }
 168     }
 169 
 170     static void testConnect() throws Exception {
 171         System.out.println("-- connect --");
 172 
 173         SocketAddress address;
 174 
 175         try (Server server = new Server()) {
 176             address = server.address();
 177 
 178             // connect to server and check local/remote addresses
 179             try (AsynchronousSocketChannel ch = AsynchronousSocketChannel.open()) {
 180                 ch.connect(address).get();
 181                 // check local address
 182                 if (ch.getLocalAddress() == null)
 183                     throw new RuntimeException("Not bound to local address");
 184 
 185                 // check remote address
 186                 InetSocketAddress remote = (InetSocketAddress)ch.getRemoteAddress();
 187                 if (remote.getPort() != server.address().getPort())
 188                     throw new RuntimeException("Connected to unexpected port");
 189                 if (!remote.getAddress().equals(server.address().getAddress()))
 190                     throw new RuntimeException("Connected to unexpected address");
 191 
 192                 // try to connect again
 193                 try {
 194                     ch.connect(server.address()).get();
 195                     throw new RuntimeException("AlreadyConnectedException expected");
 196                 } catch (AlreadyConnectedException x) {
 197                 }
 198 
 199                 // clean-up
 200                 server.accept().close();
 201             }
 202 
 203             // check that connect fails with ClosedChannelException
 204             AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 205             ch.close();
 206             try {
 207                 ch.connect(server.address()).get();
 208                 throw new RuntimeException("ExecutionException expected");
 209             } catch (ExecutionException x) {
 210                 if (!(x.getCause() instanceof ClosedChannelException))
 211                     throw new RuntimeException("Cause of ClosedChannelException expected");
 212             }
 213             final AtomicReference<Throwable> connectException = new AtomicReference<>();
 214             ch.connect(server.address(), (Void)null, new CompletionHandler<Void,Void>() {
 215                 public void completed(Void result, Void att) {
 216                 }
 217                 public void failed(Throwable exc, Void att) {
 218                     connectException.set(exc);
 219                 }
 220             });
 221             while (connectException.get() == null) {
 222                 Thread.sleep(100);
 223             }
 224             if (!(connectException.get() instanceof ClosedChannelException))
 225                 throw new RuntimeException("ClosedChannelException expected");
 226         }
 227 
 228         // test that failure to connect closes the channel
 229         try (AsynchronousSocketChannel ch = AsynchronousSocketChannel.open()) {
 230             try {
 231                 ch.connect(address).get();
 232             } catch (ExecutionException x) {
 233                 // failed to establish connection
 234                 if (ch.isOpen())
 235                     throw new RuntimeException("Channel should be closed");
 236             }
 237         }
 238 
 239         // repeat test by connecting to a (probably) non-existent host. This
 240         // improves the chance that the connect will not fail immediately.
 241         if (!skipSlowConnectTest) {
 242             try (AsynchronousSocketChannel ch = AsynchronousSocketChannel.open()) {
 243                 try {
 244                     ch.connect(genSocketAddress()).get();
 245                 } catch (ExecutionException x) {
 246                     // failed to establish connection
 247                     if (ch.isOpen())
 248                         throw new RuntimeException("Channel should be closed");
 249                 }
 250             }
 251         }
 252     }
 253 
 254     static void testCloseWhenPending() throws Exception {
 255         System.out.println("-- asynchronous close when connecting --");
 256 
 257         AsynchronousSocketChannel ch;
 258 
 259         // asynchronous close while connecting
 260         ch = AsynchronousSocketChannel.open();
 261         Future<Void> connectResult = ch.connect(genSocketAddress());
 262 
 263         // give time to initiate the connect (SYN)
 264         Thread.sleep(50);
 265 
 266         // close
 267         ch.close();
 268 
 269         // check that exception is thrown in timely manner
 270         try {
 271             connectResult.get(5, TimeUnit.SECONDS);
 272         } catch (TimeoutException x) {
 273             throw new RuntimeException("AsynchronousCloseException not thrown");
 274         } catch (ExecutionException x) {
 275             // expected
 276         }
 277 
 278         System.out.println("-- asynchronous close when reading --");
 279 
 280         try (Server server = new Server()) {
 281             ch = AsynchronousSocketChannel.open();
 282             ch.connect(server.address()).get();
 283 
 284             ByteBuffer dst = ByteBuffer.allocateDirect(100);
 285             Future<Integer> result = ch.read(dst);
 286 
 287             // attempt a second read - should fail with ReadPendingException
 288             ByteBuffer buf = ByteBuffer.allocateDirect(100);
 289             try {
 290                 ch.read(buf);
 291                 throw new RuntimeException("ReadPendingException expected");
 292             } catch (ReadPendingException x) {
 293             }
 294 
 295             // close channel (should cause initial read to complete)
 296             ch.close();
 297             server.accept().close();
 298 
 299             // check that AsynchronousCloseException is thrown
 300             try {
 301                 result.get();
 302                 throw new RuntimeException("Should not read");
 303             } catch (ExecutionException x) {
 304                 if (!(x.getCause() instanceof AsynchronousCloseException))
 305                     throw new RuntimeException(x);
 306             }
 307 
 308             System.out.println("-- asynchronous close when writing --");
 309 
 310             ch = AsynchronousSocketChannel.open();
 311             ch.connect(server.address()).get();
 312 
 313             final AtomicReference<Throwable> writeException =
 314                 new AtomicReference<Throwable>();
 315 
 316             // write bytes to fill socket buffer
 317             ch.write(genBuffer(), ch, new CompletionHandler<Integer,AsynchronousSocketChannel>() {
 318                 public void completed(Integer result, AsynchronousSocketChannel ch) {
 319                     ch.write(genBuffer(), ch, this);
 320                 }
 321                 public void failed(Throwable x, AsynchronousSocketChannel ch) {
 322                     writeException.set(x);
 323                 }
 324             });
 325 
 326             // give time for socket buffer to fill up.
 327             Thread.sleep(5*1000);
 328 
 329             //  attempt a concurrent write - should fail with WritePendingException
 330             try {
 331                 ch.write(genBuffer());
 332                 throw new RuntimeException("WritePendingException expected");
 333             } catch (WritePendingException x) {
 334             }
 335 
 336             // close channel - should cause initial write to complete
 337             ch.close();
 338             server.accept().close();
 339 
 340             // wait for exception
 341             while (writeException.get() == null) {
 342                 Thread.sleep(100);
 343             }
 344             if (!(writeException.get() instanceof AsynchronousCloseException))
 345                 throw new RuntimeException("AsynchronousCloseException expected");
 346         }
 347     }
 348 
 349     static void testCancel() throws Exception {
 350         System.out.println("-- cancel --");
 351 
 352         try (Server server = new Server()) {
 353             for (int i=0; i<2; i++) {
 354                 boolean mayInterruptIfRunning = (i == 0) ? false : true;
 355 
 356                 // establish loopback connection
 357                 AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 358                 ch.connect(server.address()).get();
 359                 SocketChannel peer = server.accept();
 360 
 361                 // start read operation
 362                 ByteBuffer buf = ByteBuffer.allocate(1);
 363                 Future<Integer> res = ch.read(buf);
 364 
 365                 // cancel operation
 366                 boolean cancelled = res.cancel(mayInterruptIfRunning);
 367 
 368                 // check post-conditions
 369                 if (!res.isDone())
 370                     throw new RuntimeException("isDone should return true");
 371                 if (res.isCancelled() != cancelled)
 372                     throw new RuntimeException("isCancelled not consistent");
 373                 try {
 374                     res.get();
 375                     throw new RuntimeException("CancellationException expected");
 376                 } catch (CancellationException x) {
 377                 }
 378                 try {
 379                     res.get(1, TimeUnit.SECONDS);
 380                     throw new RuntimeException("CancellationException expected");
 381                 } catch (CancellationException x) {
 382                 }
 383 
 384                 // check that the cancel doesn't impact writing to the channel
 385                 if (!mayInterruptIfRunning) {
 386                     buf = ByteBuffer.wrap("a".getBytes());
 387                     ch.write(buf).get();
 388                 }
 389 
 390                 ch.close();
 391                 peer.close();
 392             }
 393         }
 394     }
 395 
 396     static void testRead1() throws Exception {
 397         System.out.println("-- read (1) --");
 398 
 399         try (Server server = new Server()) {
 400             final AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 401             ch.connect(server.address()).get();
 402 
 403             // read with 0 bytes remaining should complete immediately
 404             ByteBuffer buf = ByteBuffer.allocate(1);
 405             buf.put((byte)0);
 406             int n = ch.read(buf).get();
 407             if (n != 0)
 408                 throw new RuntimeException("0 expected");
 409 
 410             // write bytes and close connection
 411             ByteBuffer src = genBuffer();
 412             try (SocketChannel sc = server.accept()) {
 413                 sc.setOption(SO_SNDBUF, src.remaining());
 414                 while (src.hasRemaining())
 415                     sc.write(src);
 416             }
 417 
 418             // reads should complete immediately
 419             final ByteBuffer dst = ByteBuffer.allocateDirect(src.capacity() + 100);
 420             final CountDownLatch latch = new CountDownLatch(1);
 421             ch.read(dst, (Void)null, new CompletionHandler<Integer,Void>() {
 422                 public void completed(Integer result, Void att) {
 423                     int n = result;
 424                     if (n > 0) {
 425                         ch.read(dst, (Void)null, this);
 426                     } else {
 427                         latch.countDown();
 428                     }
 429                 }
 430                 public void failed(Throwable exc, Void att) {
 431                 }
 432             });
 433 
 434             latch.await();
 435 
 436             // check buffers
 437             src.flip();
 438             dst.flip();
 439             if (!src.equals(dst)) {
 440                 throw new RuntimeException("Contents differ");
 441             }
 442 
 443             // close channel
 444             ch.close();
 445 
 446             // check read fails with ClosedChannelException
 447             try {
 448                 ch.read(dst).get();
 449                 throw new RuntimeException("ExecutionException expected");
 450             } catch (ExecutionException x) {
 451                 if (!(x.getCause() instanceof ClosedChannelException))
 452                     throw new RuntimeException("Cause of ClosedChannelException expected");
 453             }
 454         }
 455     }
 456 
 457     static void testRead2() throws Exception {
 458         System.out.println("-- read (2) --");
 459 
 460         try (Server server = new Server()) {
 461             final AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 462             ch.connect(server.address()).get();
 463             SocketChannel sc = server.accept();
 464 
 465             ByteBuffer src = genBuffer();
 466 
 467             // read until the buffer is full
 468             final ByteBuffer dst = ByteBuffer.allocateDirect(src.capacity());
 469             final CountDownLatch latch = new CountDownLatch(1);
 470             ch.read(dst, (Void)null, new CompletionHandler<Integer,Void>() {
 471                 public void completed(Integer result, Void att) {
 472                     if (dst.hasRemaining()) {
 473                         ch.read(dst, (Void)null, this);
 474                     } else {
 475                         latch.countDown();
 476                     }
 477                 }
 478                 public void failed(Throwable exc, Void att) {
 479                 }
 480             });
 481 
 482             // trickle the writing
 483             do {
 484                 int rem = src.remaining();
 485                 int size = (rem <= 100) ? rem : 50 + rand.nextInt(rem - 100);
 486                 ByteBuffer buf = ByteBuffer.allocate(size);
 487                 for (int i=0; i<size; i++)
 488                     buf.put(src.get());
 489                 buf.flip();
 490                 Thread.sleep(50 + rand.nextInt(1500));
 491                 while (buf.hasRemaining())
 492                     sc.write(buf);
 493             } while (src.hasRemaining());
 494 
 495             // wait until ascynrhonous reading has completed
 496             latch.await();
 497 
 498             // check buffers
 499             src.flip();
 500             dst.flip();
 501             if (!src.equals(dst)) {
 502                throw new RuntimeException("Contents differ");
 503             }
 504 
 505             sc.close();
 506             ch.close();
 507         }
 508     }
 509 
 510     // exercise scattering read
 511     static void testRead3() throws Exception {
 512         System.out.println("-- read (3) --");
 513 
 514         try (Server server = new Server()) {
 515             final AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 516             ch.connect(server.address()).get();
 517             SocketChannel sc = server.accept();
 518 
 519             ByteBuffer[] dsts = new ByteBuffer[3];
 520             for (int i=0; i<dsts.length; i++) {
 521                 dsts[i] = ByteBuffer.allocateDirect(100);
 522             }
 523 
 524             // scattering read that completes ascynhronously
 525             final CountDownLatch l1 = new CountDownLatch(1);
 526             ch.read(dsts, 0, dsts.length, 0L, TimeUnit.SECONDS, (Void)null,
 527                 new CompletionHandler<Long,Void>() {
 528                     public void completed(Long result, Void att) {
 529                         long n = result;
 530                         if (n <= 0)
 531                             throw new RuntimeException("No bytes read");
 532                         l1.countDown();
 533                     }
 534                     public void failed(Throwable exc, Void att) {
 535                     }
 536             });
 537 
 538             // write some bytes
 539             sc.write(genBuffer());
 540 
 541             // read should now complete
 542             l1.await();
 543 
 544             // write more bytes
 545             sc.write(genBuffer());
 546 
 547             // read should complete immediately
 548             for (int i=0; i<dsts.length; i++) {
 549                 dsts[i].rewind();
 550             }
 551 
 552             final CountDownLatch l2 = new CountDownLatch(1);
 553             ch.read(dsts, 0, dsts.length, 0L, TimeUnit.SECONDS, (Void)null,
 554                 new CompletionHandler<Long,Void>() {
 555                     public void completed(Long result, Void att) {
 556                         long n = result;
 557                         if (n <= 0)
 558                             throw new RuntimeException("No bytes read");
 559                         l2.countDown();
 560                     }
 561                     public void failed(Throwable exc, Void att) {
 562                     }
 563             });
 564             l2.await();
 565 
 566             ch.close();
 567             sc.close();
 568         }
 569     }
 570 
 571     static void testWrite1() throws Exception {
 572         System.out.println("-- write (1) --");
 573 
 574         try (Server server = new Server()) {
 575             final AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 576             ch.connect(server.address()).get();
 577             SocketChannel sc = server.accept();
 578 
 579             // write with 0 bytes remaining should complete immediately
 580             ByteBuffer buf = ByteBuffer.allocate(1);
 581             buf.put((byte)0);
 582             int n = ch.write(buf).get();
 583             if (n != 0)
 584                 throw new RuntimeException("0 expected");
 585 
 586             // write all bytes and close connection when done
 587             final ByteBuffer src = genBuffer();
 588             ch.write(src, (Void)null, new CompletionHandler<Integer,Void>() {
 589                 public void completed(Integer result, Void att) {
 590                     if (src.hasRemaining()) {
 591                         ch.write(src, (Void)null, this);
 592                     } else {
 593                         try {
 594                             ch.close();
 595                         } catch (IOException ignore) { }
 596                     }
 597                 }
 598                 public void failed(Throwable exc, Void att) {
 599                 }
 600             });
 601 
 602             // read to EOF or buffer full
 603             ByteBuffer dst = ByteBuffer.allocateDirect(src.capacity() + 100);
 604             do {
 605                 n = sc.read(dst);
 606             } while (n > 0);
 607             sc.close();
 608 
 609             // check buffers
 610             src.flip();
 611             dst.flip();
 612             if (!src.equals(dst)) {
 613                 throw new RuntimeException("Contents differ");
 614             }
 615 
 616             // check write fails with ClosedChannelException
 617             try {
 618                 ch.read(dst).get();
 619                 throw new RuntimeException("ExecutionException expected");
 620             } catch (ExecutionException x) {
 621                 if (!(x.getCause() instanceof ClosedChannelException))
 622                     throw new RuntimeException("Cause of ClosedChannelException expected");
 623             }
 624         }
 625     }
 626 
 627     // exercise gathering write
 628     static void testWrite2() throws Exception {
 629         System.out.println("-- write (2) --");
 630 
 631         try (Server server = new Server()) {
 632             final AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 633             ch.connect(server.address()).get();
 634             SocketChannel sc = server.accept();
 635 
 636             // number of bytes written
 637             final AtomicLong bytesWritten = new AtomicLong(0);
 638 
 639             // write buffers (should complete immediately)
 640             ByteBuffer[] srcs = genBuffers(1);
 641             final CountDownLatch l1 = new CountDownLatch(1);
 642             ch.write(srcs, 0, srcs.length, 0L, TimeUnit.SECONDS, (Void)null,
 643                 new CompletionHandler<Long,Void>() {
 644                     public void completed(Long result, Void att) {
 645                         long n = result;
 646                         if (n <= 0)
 647                             throw new RuntimeException("No bytes read");
 648                         bytesWritten.addAndGet(n);
 649                         l1.countDown();
 650                     }
 651                     public void failed(Throwable exc, Void att) {
 652                     }
 653             });
 654             l1.await();
 655 
 656             // set to true to signal that no more buffers should be written
 657             final AtomicBoolean continueWriting = new AtomicBoolean(true);
 658 
 659             // write until socket buffer is full so as to create the conditions
 660             // for when a write does not complete immediately
 661             srcs = genBuffers(1);
 662             ch.write(srcs, 0, srcs.length, 0L, TimeUnit.SECONDS, (Void)null,
 663                 new CompletionHandler<Long,Void>() {
 664                     public void completed(Long result, Void att) {
 665                         long n = result;
 666                         if (n <= 0)
 667                             throw new RuntimeException("No bytes written");
 668                         bytesWritten.addAndGet(n);
 669                         if (continueWriting.get()) {
 670                             ByteBuffer[] srcs = genBuffers(8);
 671                             ch.write(srcs, 0, srcs.length, 0L, TimeUnit.SECONDS,
 672                                 (Void)null, this);
 673                         }
 674                     }
 675                     public void failed(Throwable exc, Void att) {
 676                     }
 677             });
 678 
 679             // give time for socket buffer to fill up.
 680             Thread.sleep(5*1000);
 681 
 682             // signal handler to stop further writing
 683             continueWriting.set(false);
 684 
 685             // read until done
 686             ByteBuffer buf = ByteBuffer.allocateDirect(4096);
 687             long total = 0L;
 688             do {
 689                 int n = sc.read(buf);
 690                 if (n <= 0)
 691                     throw new RuntimeException("No bytes read");
 692                 buf.rewind();
 693                 total += n;
 694             } while (total < bytesWritten.get());
 695 
 696             ch.close();
 697             sc.close();
 698         }
 699     }
 700 
 701     static void testShutdown() throws Exception {
 702         System.out.println("-- shutdown--");
 703 
 704         try (Server server = new Server();
 705              AsynchronousSocketChannel ch = AsynchronousSocketChannel.open())
 706         {
 707             ch.connect(server.address()).get();
 708             try (SocketChannel peer = server.accept()) {
 709                 ByteBuffer buf = ByteBuffer.allocateDirect(1000);
 710                 int n;
 711 
 712                 // check read
 713                 ch.shutdownInput();
 714                 n = ch.read(buf).get();
 715                 if (n != -1)
 716                     throw new RuntimeException("-1 expected");
 717                 // check full with full buffer
 718                 buf.put(new byte[100]);
 719                 n = ch.read(buf).get();
 720                 if (n != -1)
 721                     throw new RuntimeException("-1 expected");
 722 
 723                 // check write
 724                 ch.shutdownOutput();
 725                 try {
 726                     ch.write(buf).get();
 727                     throw new RuntimeException("ClosedChannelException expected");
 728                 } catch (ExecutionException x) {
 729                     if (!(x.getCause() instanceof ClosedChannelException))
 730                         throw new RuntimeException("ClosedChannelException expected");
 731                 }
 732             }
 733         }
 734     }
 735 
 736     static void testTimeout() throws Exception {
 737         System.out.println("-- timeouts --");
 738         testTimeout(Integer.MIN_VALUE, TimeUnit.SECONDS);
 739         testTimeout(-1L, TimeUnit.SECONDS);
 740         testTimeout(0L, TimeUnit.SECONDS);
 741         testTimeout(2L, TimeUnit.SECONDS);
 742     }
 743 
 744     static void testTimeout(final long timeout, final TimeUnit unit) throws Exception {
 745         try (Server server = new Server()) {
 746             AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
 747             ch.connect(server.address()).get();
 748 
 749             ByteBuffer dst = ByteBuffer.allocate(512);
 750 
 751             final AtomicReference<Throwable> readException = new AtomicReference<Throwable>();
 752 
 753             // this read should timeout if value is > 0
 754             ch.read(dst, timeout, unit, null, new CompletionHandler<Integer,Void>() {
 755                 public void completed(Integer result, Void att) {
 756                     readException.set(new RuntimeException("Should not complete"));
 757                 }
 758                 public void failed(Throwable exc, Void att) {
 759                     readException.set(exc);
 760                 }
 761             });
 762             if (timeout > 0L) {
 763                 // wait for exception
 764                 while (readException.get() == null) {
 765                     Thread.sleep(100);
 766                 }
 767                 if (!(readException.get() instanceof InterruptedByTimeoutException))
 768                     throw new RuntimeException("InterruptedByTimeoutException expected");
 769 
 770                 // after a timeout then further reading should throw unspecified runtime exception
 771                 boolean exceptionThrown = false;
 772                 try {
 773                     ch.read(dst);
 774                 } catch (RuntimeException x) {
 775                     exceptionThrown = true;
 776                 }
 777                 if (!exceptionThrown)
 778                     throw new RuntimeException("RuntimeException expected after timeout.");
 779             } else {
 780                 Thread.sleep(1000);
 781                 Throwable exc = readException.get();
 782                 if (exc != null)
 783                     throw new RuntimeException(exc);
 784             }
 785 
 786             final AtomicReference<Throwable> writeException = new AtomicReference<Throwable>();
 787 
 788             // write bytes to fill socket buffer
 789             ch.write(genBuffer(), timeout, unit, ch,
 790                 new CompletionHandler<Integer,AsynchronousSocketChannel>()
 791             {
 792                 public void completed(Integer result, AsynchronousSocketChannel ch) {
 793                     ch.write(genBuffer(), timeout, unit, ch, this);
 794                 }
 795                 public void failed(Throwable exc, AsynchronousSocketChannel ch) {
 796                     writeException.set(exc);
 797                 }
 798             });
 799             if (timeout > 0) {
 800                 // wait for exception
 801                 while (writeException.get() == null) {
 802                     Thread.sleep(100);
 803                 }
 804                 if (!(writeException.get() instanceof InterruptedByTimeoutException))
 805                     throw new RuntimeException("InterruptedByTimeoutException expected");
 806 
 807                 // after a timeout then further writing should throw unspecified runtime exception
 808                 boolean exceptionThrown = false;
 809                 try {
 810                     ch.write(genBuffer());
 811                 } catch (RuntimeException x) {
 812                     exceptionThrown = true;
 813                 }
 814                 if (!exceptionThrown)
 815                     throw new RuntimeException("RuntimeException expected after timeout.");
 816             } else {
 817                 Thread.sleep(1000);
 818                 Throwable exc = writeException.get();
 819                 if (exc != null)
 820                     throw new RuntimeException(exc);
 821             }
 822 
 823             // clean-up
 824             server.accept().close();
 825             ch.close();
 826         }
 827     }
 828 
 829     // returns ByteBuffer with random bytes
 830     static ByteBuffer genBuffer() {
 831         int size = 1024 + rand.nextInt(16000);
 832         byte[] buf = new byte[size];
 833         rand.nextBytes(buf);
 834         boolean useDirect = rand.nextBoolean();
 835         if (useDirect) {
 836             ByteBuffer bb = ByteBuffer.allocateDirect(buf.length);
 837             bb.put(buf);
 838             bb.flip();
 839             return bb;
 840         } else {
 841             return ByteBuffer.wrap(buf);
 842         }
 843     }
 844 
 845     // return ByteBuffer[] with random bytes
 846     static ByteBuffer[] genBuffers(int max) {
 847         int len = 1;
 848         if (max > 1)
 849             len += rand.nextInt(max);
 850         ByteBuffer[] bufs = new ByteBuffer[len];
 851         for (int i=0; i<len; i++)
 852             bufs[i] = genBuffer();
 853         return bufs;
 854     }
 855 
 856     // return random SocketAddress
 857     static SocketAddress genSocketAddress() {
 858         StringBuilder sb = new StringBuilder("10.");
 859         sb.append(rand.nextInt(256));
 860         sb.append('.');
 861         sb.append(rand.nextInt(256));
 862         sb.append('.');
 863         sb.append(rand.nextInt(256));
 864         InetAddress rh;
 865         try {
 866             rh = InetAddress.getByName(sb.toString());
 867         } catch (UnknownHostException x) {
 868             throw new InternalError("Should not happen");
 869         }
 870         return new InetSocketAddress(rh, rand.nextInt(65535)+1);
 871     }
 872 }
--- EOF ---