1 /*
   2  * Copyright (c) 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 /* @test
  25  * @bug 8199433
  26  * @run testng SelectWithConsumer
  27  * @summary Unit test for Selector select(Consumer), select(Consumer,long) and
  28  *          selectNow(Consumer)
  29  */
  30 
  31 import java.io.Closeable;
  32 import java.io.IOException;
  33 import java.net.InetSocketAddress;
  34 import java.nio.ByteBuffer;
  35 import java.nio.channels.ClosedSelectorException;
  36 import java.nio.channels.Pipe;
  37 import java.nio.channels.SelectionKey;
  38 import java.nio.channels.Selector;
  39 import java.nio.channels.ServerSocketChannel;
  40 import java.nio.channels.SocketChannel;
  41 import java.nio.channels.WritableByteChannel;
  42 import java.util.concurrent.Executors;
  43 import java.util.concurrent.ScheduledExecutorService;
  44 import java.util.concurrent.TimeUnit;
  45 import java.util.concurrent.atomic.AtomicInteger;
  46 import static java.util.concurrent.TimeUnit.*;
  47 
  48 import org.testng.annotations.AfterTest;
  49 import org.testng.annotations.Test;
  50 import static org.testng.Assert.*;
  51 
  52 @Test
  53 public class SelectWithConsumer {
  54 
  55     /**
  56      * Invoke the select methods that take an action and check that the
  57      * accumulated ready ops notified to the action matches the expected ops.
  58      */
  59     void testActionInvoked(SelectionKey key, int expectedOps) throws Exception {
  60         var callerThread = Thread.currentThread();
  61         var sel = key.selector();
  62         var interestOps = key.interestOps();
  63         var notifiedOps = new AtomicInteger();
  64 
  65         // select(Consumer)
  66         if (expectedOps == 0)
  67             sel.wakeup(); // ensure select does not block
  68         notifiedOps.set(0);
  69         int n = sel.select(k -> {
  70             assertTrue(Thread.currentThread() == callerThread);
  71             assertTrue(k == key);
  72             int readyOps = key.readyOps();
  73             assertTrue((readyOps & interestOps) != 0);
  74             assertTrue((readyOps & notifiedOps.get()) == 0);
  75             notifiedOps.set(notifiedOps.get() | readyOps);
  76         });
  77         assertTrue((n == 1) ^ (expectedOps == 0));
  78         assertTrue(notifiedOps.get() == expectedOps);
  79 
  80         // select(Consumer, timeout)
  81         notifiedOps.set(0);
  82         n = sel.select(k -> {
  83             assertTrue(Thread.currentThread() == callerThread);
  84             assertTrue(k == key);
  85             int readyOps = key.readyOps();
  86             assertTrue((readyOps & interestOps) != 0);
  87             assertTrue((readyOps & notifiedOps.get()) == 0);
  88             notifiedOps.set(notifiedOps.get() | readyOps);
  89         }, 1000);
  90         assertTrue((n == 1) ^ (expectedOps == 0));
  91         assertTrue(notifiedOps.get() == expectedOps);
  92 
  93         // selectNow(Consumer)
  94         notifiedOps.set(0);
  95         n = sel.selectNow(k -> {
  96             assertTrue(Thread.currentThread() == callerThread);
  97             assertTrue(k == key);
  98             int readyOps = key.readyOps();
  99             assertTrue((readyOps & interestOps) != 0);
 100             assertTrue((readyOps & notifiedOps.get()) == 0);
 101             notifiedOps.set(notifiedOps.get() | readyOps);
 102         });
 103         assertTrue((n == 1) ^ (expectedOps == 0));
 104         assertTrue(notifiedOps.get() == expectedOps);
 105     }
 106 
 107     /**
 108      * Test that an action is performed when a channel is ready for reading.
 109      */
 110     public void testReadable() throws Exception {
 111         Pipe p = Pipe.open();
 112         try (Selector sel = Selector.open()) {
 113             Pipe.SinkChannel sink = p.sink();
 114             Pipe.SourceChannel source = p.source();
 115             source.configureBlocking(false);
 116             SelectionKey key = source.register(sel, SelectionKey.OP_READ);
 117 
 118             // write to sink to ensure source is readable
 119             scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
 120 
 121             // test that action is invoked
 122             testActionInvoked(key, SelectionKey.OP_READ);
 123         } finally {
 124             closePipe(p);
 125         }
 126     }
 127 
 128     /**
 129      * Test that an action is performed when a channel is ready for writing.
 130      */
 131     public void testWritable() throws Exception {
 132         Pipe p = Pipe.open();
 133         try (Selector sel = Selector.open()) {
 134             Pipe.SourceChannel source = p.source();
 135             Pipe.SinkChannel sink = p.sink();
 136             sink.configureBlocking(false);
 137             SelectionKey key = sink.register(sel, SelectionKey.OP_WRITE);
 138 
 139             // test that action is invoked
 140             testActionInvoked(key, SelectionKey.OP_WRITE);
 141         } finally {
 142             closePipe(p);
 143         }
 144     }
 145 
 146     /**
 147      * Test that an action is performed when a channel is ready for both
 148      * reading and writing.
 149      */
 150     public void testReadableAndWriteable() throws Exception {
 151         ServerSocketChannel ssc = null;
 152         SocketChannel sc = null;
 153         SocketChannel peer = null;
 154         try (Selector sel = Selector.open()) {
 155             ssc = ServerSocketChannel.open().bind(new InetSocketAddress(0));
 156             sc = SocketChannel.open(ssc.getLocalAddress());
 157             sc.configureBlocking(false);
 158             SelectionKey key = sc.register(sel, (SelectionKey.OP_READ |
 159                                                  SelectionKey.OP_WRITE));
 160 
 161             // accept connection and write data so the source is readable
 162             peer = ssc.accept();
 163             peer.write(messageBuffer());
 164 
 165             // test that action is invoked
 166             testActionInvoked(key, (SelectionKey.OP_READ | SelectionKey.OP_WRITE));
 167         } finally {
 168             if (ssc != null) ssc.close();
 169             if (sc != null) sc.close();
 170             if (peer != null) peer.close();
 171         }
 172     }
 173 
 174     /**
 175      * Test that the action is called for two selected channels
 176      */
 177     public void testTwoChannels() throws Exception {
 178         Pipe p = Pipe.open();
 179         try (Selector sel = Selector.open()) {
 180             Pipe.SourceChannel source = p.source();
 181             Pipe.SinkChannel sink = p.sink();
 182             source.configureBlocking(false);
 183             sink.configureBlocking(false);
 184             SelectionKey key1 = source.register(sel, SelectionKey.OP_READ);
 185             SelectionKey key2 = sink.register(sel, SelectionKey.OP_WRITE);
 186 
 187             // write to sink to ensure that the source is readable
 188             sink.write(messageBuffer());
 189 
 190             var counter = new AtomicInteger();
 191 
 192             // select(Consumer)
 193             counter.set(0);
 194             int n = sel.select(k -> {
 195                 counter.incrementAndGet();
 196                 if (k == key1) {
 197                     assertTrue(k.isReadable());
 198                 } else if (k == key2) {
 199                     assertTrue(k.isWritable());
 200                 } else {
 201                     assertTrue(false);
 202                 }
 203             });
 204             assertTrue(n == 2);
 205             assertTrue(counter.get() == 2);
 206 
 207             // select(Consumer, timeout)
 208             counter.set(0);
 209             n = sel.select(k -> {
 210                 counter.incrementAndGet();
 211                 if (k == key1) {
 212                     assertTrue(k.isReadable());
 213                 } else if (k == key2) {
 214                     assertTrue(k.isWritable());
 215                 } else {
 216                     assertTrue(false);
 217                 }
 218             }, 1000);
 219             assertTrue(n == 2);
 220             assertTrue(counter.get() == 2);
 221 
 222             // selectNow(Consumer)
 223             counter.set(0);
 224             n = sel.selectNow(k -> {
 225                 counter.incrementAndGet();
 226                 if (k == key1) {
 227                     assertTrue(k.isReadable());
 228                 } else if (k == key2) {
 229                     assertTrue(k.isWritable());
 230                 } else {
 231                     assertTrue(false);
 232                 }
 233             });
 234             assertTrue(n == 2);
 235             assertTrue(counter.get() == 2);
 236         } finally {
 237             closePipe(p);
 238         }
 239     }
 240 
 241     /**
 242      * Test calling select twice, the action should be invoked each time
 243      */
 244     public void testRepeatedSelect1() throws Exception {
 245         Pipe p = Pipe.open();
 246         try (Selector sel = Selector.open()) {
 247             Pipe.SourceChannel source = p.source();
 248             Pipe.SinkChannel sink = p.sink();
 249             source.configureBlocking(false);
 250             SelectionKey key = source.register(sel, SelectionKey.OP_READ);
 251 
 252             // write to sink to ensure that the source is readable
 253             sink.write(messageBuffer());
 254 
 255             // test that action is invoked
 256             testActionInvoked(key, SelectionKey.OP_READ);
 257             testActionInvoked(key, SelectionKey.OP_READ);
 258 
 259         } finally {
 260             closePipe(p);
 261         }
 262     }
 263 
 264     /**
 265      * Test calling select twice. An I/O operation is performed after the
 266      * first select so the channel will not be selected by the second select.
 267      */
 268     public void testRepeatedSelect2() throws Exception {
 269         Pipe p = Pipe.open();
 270         try (Selector sel = Selector.open()) {
 271             Pipe.SourceChannel source = p.source();
 272             Pipe.SinkChannel sink = p.sink();
 273             source.configureBlocking(false);
 274             SelectionKey key = source.register(sel, SelectionKey.OP_READ);
 275 
 276             // write to sink to ensure that the source is readable
 277             sink.write(messageBuffer());
 278 
 279             // test that action is invoked
 280             testActionInvoked(key, SelectionKey.OP_READ);
 281 
 282             // read all bytes
 283             int n;
 284             ByteBuffer bb = ByteBuffer.allocate(100);
 285             do {
 286                 n = source.read(bb);
 287                 bb.clear();
 288             } while (n > 0);
 289 
 290             // test that action is not invoked
 291             testActionInvoked(key, 0);
 292         } finally {
 293             closePipe(p);
 294         }
 295     }
 296 
 297     /**
 298      * Test timeout
 299      */
 300     public void testTimeout() throws Exception {
 301         Pipe p = Pipe.open();
 302         try (Selector sel = Selector.open()) {
 303             Pipe.SourceChannel source = p.source();
 304             Pipe.SinkChannel sink = p.sink();
 305             source.configureBlocking(false);
 306             source.register(sel, SelectionKey.OP_READ);
 307             long start = System.currentTimeMillis();
 308             int n = sel.select(k -> assertTrue(false), 1000L);
 309             long duration = System.currentTimeMillis() - start;
 310             assertTrue(n == 0);
 311             assertTrue(duration > 500, "select took " + duration + " ms");
 312         } finally {
 313             closePipe(p);
 314         }
 315     }
 316 
 317     /**
 318      * Test wakeup prior to select
 319      */
 320     public void testWakeupBeforeSelect() throws Exception {
 321         // select(Consumer)
 322         try (Selector sel = Selector.open()) {
 323             sel.wakeup();
 324             int n = sel.select(k -> assertTrue(false));
 325             assertTrue(n == 0);
 326         }
 327 
 328         // select(Consumer, timeout)
 329         try (Selector sel = Selector.open()) {
 330             sel.wakeup();
 331             long start = System.currentTimeMillis();
 332             int n = sel.select(k -> assertTrue(false), 60*1000);
 333             long duration = System.currentTimeMillis() - start;
 334             assertTrue(n == 0);
 335             assertTrue(duration < 5000, "select took " + duration + " ms");
 336         }
 337     }
 338 
 339     /**
 340      * Test wakeup during select
 341      */
 342     public void testWakeupDuringSelect() throws Exception {
 343         // select(Consumer)
 344         try (Selector sel = Selector.open()) {
 345             scheduleWakeup(sel, 1, SECONDS);
 346             int n = sel.select(k -> assertTrue(false));
 347             assertTrue(n == 0);
 348         }
 349 
 350         // select(Consumer, timeout)
 351         try (Selector sel = Selector.open()) {
 352             scheduleWakeup(sel, 1, SECONDS);
 353             long start = System.currentTimeMillis();
 354             int n = sel.select(k -> assertTrue(false), 60*1000);
 355             long duration = System.currentTimeMillis() - start;
 356             assertTrue(n == 0);
 357             assertTrue(duration > 500 && duration < 10*1000,
 358                     "select took " + duration + " ms");
 359         }
 360     }
 361 
 362     /**
 363      * Test invoking select with interrupt status set
 364      */
 365     public void testInterruptBeforeSelect() throws Exception {
 366         // select(Consumer)
 367         try (Selector sel = Selector.open()) {
 368             Thread.currentThread().interrupt();
 369             int n = sel.select(k -> assertTrue(false));
 370             assertTrue(n == 0);
 371             assertTrue(Thread.currentThread().isInterrupted());
 372             assertTrue(sel.isOpen());
 373         } finally {
 374             Thread.currentThread().interrupted();  // clear interrupt status
 375         }
 376 
 377         // select(Consumer, timeout)
 378         try (Selector sel = Selector.open()) {
 379             Thread.currentThread().interrupt();
 380             long start = System.currentTimeMillis();
 381             int n = sel.select(k -> assertTrue(false), 60*1000);
 382             long duration = System.currentTimeMillis() - start;
 383             assertTrue(n == 0);
 384             assertTrue(duration < 5000, "select took " + duration + " ms");
 385             assertTrue(Thread.currentThread().isInterrupted());
 386             assertTrue(sel.isOpen());
 387         } finally {
 388             Thread.currentThread().interrupted();  // clear interrupt status
 389         }
 390     }
 391 
 392     /**
 393      * Test interrupt thread during select
 394      */
 395     public void testInterruptDuringSelect() throws Exception {
 396         // select(Consumer)
 397         try (Selector sel = Selector.open()) {
 398             scheduleInterrupt(Thread.currentThread(), 1, SECONDS);
 399             int n = sel.select(k -> assertTrue(false));
 400             assertTrue(n == 0);
 401             assertTrue(Thread.currentThread().isInterrupted());
 402             assertTrue(sel.isOpen());
 403         } finally {
 404             Thread.currentThread().interrupted();  // clear interrupt status
 405         }
 406 
 407         // select(Consumer, timeout)
 408         try (Selector sel = Selector.open()) {
 409             scheduleInterrupt(Thread.currentThread(), 1, SECONDS);
 410             long start = System.currentTimeMillis();
 411             int n = sel.select(k -> assertTrue(false), 60*1000);
 412             long duration = System.currentTimeMillis() - start;
 413             assertTrue(n == 0);
 414             assertTrue(duration > 500 && duration < 5000,
 415                     "select took " + duration + " ms");
 416             assertTrue(Thread.currentThread().isInterrupted());
 417             assertTrue(sel.isOpen());
 418         } finally {
 419             Thread.currentThread().interrupted();  // clear interrupt status
 420         }
 421     }
 422 
 423     /**
 424      * Test invoking select on a closed selector
 425      */
 426     @Test(expectedExceptions = ClosedSelectorException.class)
 427     public void testClosedSelector1() throws Exception {
 428         Selector sel = Selector.open();
 429         sel.close();
 430         sel.select(k -> assertTrue(false));
 431     }
 432     @Test(expectedExceptions = ClosedSelectorException.class)
 433     public void testClosedSelector2() throws Exception {
 434         Selector sel = Selector.open();
 435         sel.close();
 436         sel.select(k -> assertTrue(false), 1000);
 437     }
 438     @Test(expectedExceptions = ClosedSelectorException.class)
 439     public void testClosedSelector3() throws Exception {
 440         Selector sel = Selector.open();
 441         sel.close();
 442         sel.selectNow(k -> assertTrue(false));
 443     }
 444 
 445     /**
 446      * Test closing selector while in a selection operation
 447      */
 448     public void testCloseDuringSelect() throws Exception {
 449         // select(Consumer)
 450         try (Selector sel = Selector.open()) {
 451             scheduleClose(sel, 3, SECONDS);
 452             int n = sel.select(k -> assertTrue(false));
 453             assertTrue(n == 0);
 454             assertFalse(sel.isOpen());
 455         }
 456 
 457         // select(Consumer, timeout)
 458         try (Selector sel = Selector.open()) {
 459             scheduleClose(sel, 3, SECONDS);
 460             long start = System.currentTimeMillis();
 461             int n = sel.select(k -> assertTrue(false), 60*1000);
 462             long duration = System.currentTimeMillis() - start;
 463             assertTrue(n == 0);
 464             assertTrue(duration > 2000 && duration < 10*1000,
 465                     "select took " + duration + " ms");
 466             assertFalse(sel.isOpen());
 467         }
 468     }
 469 
 470     /**
 471      * Test action closing selector
 472      */
 473     @Test(expectedExceptions = ClosedSelectorException.class)
 474     public void testActionClosingSelector() throws Exception {
 475         Pipe p = Pipe.open();
 476         try (Selector sel = Selector.open()) {
 477             Pipe.SourceChannel source = p.source();
 478             Pipe.SinkChannel sink = p.sink();
 479             source.configureBlocking(false);
 480             SelectionKey key = source.register(sel, SelectionKey.OP_READ);
 481 
 482             // write to sink to ensure that the source is readable
 483             sink.write(messageBuffer());
 484 
 485             // should relay ClosedSelectorException
 486             sel.select(k -> {
 487                 assertTrue(k == key);
 488                 try {
 489                     sel.close();
 490                 } catch (IOException ioe) { }
 491             });
 492         } finally {
 493             closePipe(p);
 494         }
 495     }
 496 
 497     /**
 498      * Test that the action is invoked while synchronized on the selector and
 499      * its selected-key set.
 500      */
 501     public void testLocks() throws Exception {
 502         Pipe p = Pipe.open();
 503         try (Selector sel = Selector.open()) {
 504             Pipe.SourceChannel source = p.source();
 505             Pipe.SinkChannel sink = p.sink();
 506             source.configureBlocking(false);
 507             SelectionKey key = source.register(sel, SelectionKey.OP_READ);
 508 
 509             // write to sink to ensure that the source is readable
 510             sink.write(messageBuffer());
 511 
 512             // select(Consumer)
 513             sel.select(k -> {
 514                 assertTrue(k == key);
 515                 assertTrue(Thread.holdsLock(sel));
 516                 assertFalse(Thread.holdsLock(sel.keys()));
 517                 assertTrue(Thread.holdsLock(sel.selectedKeys()));
 518             });
 519 
 520             // select(Consumer, timeout)
 521             sel.select(k -> {
 522                 assertTrue(k == key);
 523                 assertTrue(Thread.holdsLock(sel));
 524                 assertFalse(Thread.holdsLock(sel.keys()));
 525                 assertTrue(Thread.holdsLock(sel.selectedKeys()));
 526             }, 1000L);
 527 
 528             // selectNow(Consumer)
 529             sel.selectNow(k -> {
 530                 assertTrue(k == key);
 531                 assertTrue(Thread.holdsLock(sel));
 532                 assertFalse(Thread.holdsLock(sel.keys()));
 533                 assertTrue(Thread.holdsLock(sel.selectedKeys()));
 534             });
 535         } finally {
 536             closePipe(p);
 537         }
 538     }
 539 
 540     /**
 541      * Test that selection operations remove cancelled keys from the selector's
 542      * key and selected-key sets.
 543      */
 544     public void testCancel() throws Exception {
 545         Pipe p = Pipe.open();
 546         try (Selector sel = Selector.open()) {
 547             Pipe.SinkChannel sink = p.sink();
 548             Pipe.SourceChannel source = p.source();
 549 
 550             // write to sink to ensure that the source is readable
 551             sink.write(messageBuffer());
 552 
 553             sink.configureBlocking(false);
 554             source.configureBlocking(false);
 555             SelectionKey key1 = sink.register(sel, SelectionKey.OP_WRITE);
 556             SelectionKey key2 = source.register(sel, SelectionKey.OP_READ);
 557 
 558             sel.selectNow();
 559             assertTrue(sel.keys().contains(key1));
 560             assertTrue(sel.keys().contains(key2));
 561             assertTrue(sel.selectedKeys().contains(key1));
 562             assertTrue(sel.selectedKeys().contains(key2));
 563 
 564             // cancel key1
 565             key1.cancel();
 566             int n = sel.selectNow(k -> assertTrue(k == key2));
 567             assertTrue(n == 1);
 568             assertFalse(sel.keys().contains(key1));
 569             assertTrue(sel.keys().contains(key2));
 570             assertFalse(sel.selectedKeys().contains(key1));
 571             assertTrue(sel.selectedKeys().contains(key2));
 572 
 573             // cancel key2
 574             key2.cancel();
 575             n = sel.selectNow(k -> assertTrue(false));
 576             assertTrue(n == 0);
 577             assertFalse(sel.keys().contains(key1));
 578             assertFalse(sel.keys().contains(key2));
 579             assertFalse(sel.selectedKeys().contains(key1));
 580             assertFalse(sel.selectedKeys().contains(key2));
 581         } finally {
 582             closePipe(p);
 583         }
 584     }
 585 
 586     /**
 587      * Test an action invoking select()
 588      */
 589     public void testReentrantSelect1() throws Exception {
 590         Pipe p = Pipe.open();
 591         try (Selector sel = Selector.open()) {
 592             Pipe.SinkChannel sink = p.sink();
 593             Pipe.SourceChannel source = p.source();
 594             source.configureBlocking(false);
 595             source.register(sel, SelectionKey.OP_READ);
 596 
 597             // write to sink to ensure that the source is readable
 598             scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
 599 
 600             int n = sel.select(k -> {
 601                 try {
 602                     sel.select();
 603                     assertTrue(false);
 604                 } catch (IOException ioe) {
 605                     throw new RuntimeException(ioe);
 606                 } catch (IllegalStateException expected) {
 607                 }
 608             });
 609             assertTrue(n == 1);
 610         } finally {
 611             closePipe(p);
 612         }
 613     }
 614 
 615     /**
 616      * Test an action invoking selectNow()
 617      */
 618     public void testReentrantSelect2() throws Exception {
 619         Pipe p = Pipe.open();
 620         try (Selector sel = Selector.open()) {
 621             Pipe.SinkChannel sink = p.sink();
 622             Pipe.SourceChannel source = p.source();
 623 
 624             // write to sink to ensure that the source is readable
 625             scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
 626 
 627             source.configureBlocking(false);
 628             source.register(sel, SelectionKey.OP_READ);
 629             int n = sel.select(k -> {
 630                 try {
 631                     sel.selectNow();
 632                     assertTrue(false);
 633                 } catch (IOException ioe) {
 634                     throw new RuntimeException(ioe);
 635                 } catch (IllegalStateException expected) {
 636                 }
 637             });
 638             assertTrue(n == 1);
 639         } finally {
 640             closePipe(p);
 641         }
 642     }
 643 
 644     /**
 645      * Test an action invoking select(Consumer)
 646      */
 647     public void testReentrantSelect3() throws Exception {
 648         Pipe p = Pipe.open();
 649         try (Selector sel = Selector.open()) {
 650             Pipe.SinkChannel sink = p.sink();
 651             Pipe.SourceChannel source = p.source();
 652 
 653             // write to sink to ensure that the source is readable
 654             scheduleWrite(sink, messageBuffer(), 100, MILLISECONDS);
 655 
 656             source.configureBlocking(false);
 657             source.register(sel, SelectionKey.OP_READ);
 658             int n = sel.select(k -> {
 659                 try {
 660                     sel.select(x -> assertTrue(false));
 661                     assertTrue(false);
 662                 } catch (IOException ioe) {
 663                     throw new RuntimeException(ioe);
 664                 } catch (IllegalStateException expected) {
 665                 }
 666             });
 667             assertTrue(n == 1);
 668         } finally {
 669             closePipe(p);
 670         }
 671     }
 672 
 673     /**
 674      * Negative timeout
 675      */
 676     @Test(expectedExceptions = IllegalArgumentException.class)
 677     public void testNegativeTimeout() throws Exception {
 678         try (Selector sel = Selector.open()) {
 679             sel.select(k -> { }, -1L);
 680         }
 681     }
 682 
 683     /**
 684      * Null action
 685      */
 686     @Test(expectedExceptions = NullPointerException.class)
 687     public void testNull1() throws Exception {
 688         try (Selector sel = Selector.open()) {
 689             sel.select(null);
 690         }
 691     }
 692     @Test(expectedExceptions = NullPointerException.class)
 693     public void testNull2() throws Exception {
 694         try (Selector sel = Selector.open()) {
 695             sel.select(null, 1000);
 696         }
 697     }
 698     @Test(expectedExceptions = NullPointerException.class)
 699     public void testNull3() throws Exception {
 700         try (Selector sel = Selector.open()) {
 701             sel.selectNow(null);
 702         }
 703     }
 704 
 705 
 706     // -- support methods ---
 707 
 708     private final ScheduledExecutorService POOL = Executors.newScheduledThreadPool(1);
 709 
 710     @AfterTest
 711     void shutdownThreadPool() {
 712         POOL.shutdown();
 713     }
 714 
 715     void scheduleWakeup(Selector sel, long delay, TimeUnit unit) {
 716         POOL.schedule(() -> sel.wakeup(), delay, unit);
 717     }
 718 
 719     void scheduleInterrupt(Thread t, long delay, TimeUnit unit) {
 720         POOL.schedule(() -> t.interrupt(), delay, unit);
 721     }
 722 
 723     void scheduleClose(Closeable c, long delay, TimeUnit unit) {
 724         POOL.schedule(() -> {
 725             try {
 726                 c.close();
 727             } catch (IOException ioe) {
 728                 ioe.printStackTrace();
 729             }
 730         }, delay, unit);
 731     }
 732 
 733     void scheduleWrite(WritableByteChannel sink, ByteBuffer buf, long delay, TimeUnit unit) {
 734         POOL.schedule(() -> {
 735             try {
 736                 sink.write(buf);
 737             } catch (IOException ioe) {
 738                 ioe.printStackTrace();
 739             }
 740         }, delay, unit);
 741     }
 742 
 743     static void closePipe(Pipe p) {
 744         try { p.sink().close(); } catch (IOException ignore) { }
 745         try { p.source().close(); } catch (IOException ignore) { }
 746     }
 747 
 748     static ByteBuffer messageBuffer() {
 749         try {
 750             return ByteBuffer.wrap("message".getBytes("UTF-8"));
 751         } catch (Exception e) {
 752             throw new RuntimeException(e);
 753         }
 754     }
 755 }
--- EOF ---