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