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