ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/test/tck/SynchronousQueueTest.java
(Generate patch)

Comparing jsr166/src/test/tck/SynchronousQueueTest.java (file contents):
Revision 1.5 by dl, Thu Sep 25 11:02:41 2003 UTC vs.
Revision 1.20 by jsr166, Tue Dec 1 09:48:12 2009 UTC

# Line 1 | Line 1
1   /*
2 < * Written by members of JCP JSR-166 Expert Group and released to the
3 < * public domain. Use, modify, and redistribute this code in any way
4 < * without acknowledgement. Other contributors include Andrew Wright,
5 < * Jeffrey Hayes, Pat Fischer, Mike Judd.
2 > * Written by Doug Lea with assistance from members of JCP JSR-166
3 > * Expert Group and released to the public domain, as explained at
4 > * http://creativecommons.org/licenses/publicdomain
5 > * Other contributors include Andrew Wright, Jeffrey Hayes,
6 > * Pat Fisher, Mike Judd.
7   */
8  
9   import junit.framework.*;
10   import java.util.*;
11   import java.util.concurrent.*;
12 + import static java.util.concurrent.TimeUnit.MILLISECONDS;
13   import java.io.*;
14  
15   public class SynchronousQueueTest extends JSR166TestCase {
16  
17      public static void main(String[] args) {
18 <        junit.textui.TestRunner.run (suite());  
18 >        junit.textui.TestRunner.run (suite());
19      }
20  
21      public static Test suite() {
22 <        return new TestSuite(SynchronousQueueTest.class);
22 >        return new TestSuite(SynchronousQueueTest.class);
23      }
24  
25      /**
# Line 26 | Line 28 | public class SynchronousQueueTest extend
28      public void testEmptyFull() {
29          SynchronousQueue q = new SynchronousQueue();
30          assertTrue(q.isEmpty());
31 <        assertEquals(0, q.size());
31 >        assertEquals(0, q.size());
32 >        assertEquals(0, q.remainingCapacity());
33 >        assertFalse(q.offer(zero));
34 >    }
35 >
36 >    /**
37 >     * A fair SynchronousQueue is both empty and full
38 >     */
39 >    public void testFairEmptyFull() {
40 >        SynchronousQueue q = new SynchronousQueue(true);
41 >        assertTrue(q.isEmpty());
42 >        assertEquals(0, q.size());
43          assertEquals(0, q.remainingCapacity());
44          assertFalse(q.offer(zero));
45      }
# Line 35 | Line 48 | public class SynchronousQueueTest extend
48       * offer(null) throws NPE
49       */
50      public void testOfferNull() {
51 <        try {
51 >        try {
52              SynchronousQueue q = new SynchronousQueue();
53              q.offer(null);
54              shouldThrow();
55 <        } catch (NullPointerException success) { }  
55 >        } catch (NullPointerException success) {}
56 >    }
57 >
58 >    /**
59 >     * add(null) throws NPE
60 >     */
61 >    public void testAddNull() {
62 >        try {
63 >            SynchronousQueue q = new SynchronousQueue();
64 >            q.add(null);
65 >            shouldThrow();
66 >        } catch (NullPointerException success) {}
67      }
68  
69      /**
# Line 54 | Line 78 | public class SynchronousQueueTest extend
78       * add throws ISE if no active taker
79       */
80      public void testAdd() {
81 <        try {
81 >        try {
82              SynchronousQueue q = new SynchronousQueue();
83              assertEquals(0, q.remainingCapacity());
84              q.add(one);
85              shouldThrow();
86 <        } catch (IllegalStateException success){
63 <        }  
86 >        } catch (IllegalStateException success) {}
87      }
88  
89      /**
# Line 71 | Line 94 | public class SynchronousQueueTest extend
94              SynchronousQueue q = new SynchronousQueue();
95              q.addAll(null);
96              shouldThrow();
97 <        }
98 <        catch (NullPointerException success) {}
97 >        } catch (NullPointerException success) {}
98 >    }
99 >
100 >    /**
101 >     * addAll(this) throws IAE
102 >     */
103 >    public void testAddAllSelf() {
104 >        try {
105 >            SynchronousQueue q = new SynchronousQueue();
106 >            q.addAll(q);
107 >            shouldThrow();
108 >        } catch (IllegalArgumentException success) {}
109      }
110 +
111      /**
112       * addAll of a collection with null elements throws NPE
113       */
# Line 83 | Line 117 | public class SynchronousQueueTest extend
117              Integer[] ints = new Integer[1];
118              q.addAll(Arrays.asList(ints));
119              shouldThrow();
120 <        }
87 <        catch (NullPointerException success) {}
120 >        } catch (NullPointerException success) {}
121      }
122      /**
123       * addAll throws ISE if no active taker
# Line 97 | Line 130 | public class SynchronousQueueTest extend
130                  ints[i] = new Integer(i);
131              q.addAll(Arrays.asList(ints));
132              shouldThrow();
133 <        }
101 <        catch (IllegalStateException success) {}
133 >        } catch (IllegalStateException success) {}
134      }
135  
136      /**
137       * put(null) throws NPE
138       */
139 <    public void testPutNull() {
140 <        try {
139 >    public void testPutNull() throws InterruptedException {
140 >        try {
141              SynchronousQueue q = new SynchronousQueue();
142              q.put(null);
143              shouldThrow();
144 <        }
113 <        catch (NullPointerException success){
114 <        }  
115 <        catch (InterruptedException ie) {
116 <            unexpectedException();
117 <        }
144 >        } catch (NullPointerException success) {}
145       }
146  
147      /**
148       * put blocks interruptibly if no active taker
149       */
150 <    public void testBlockingPut() {
151 <        Thread t = new Thread(new Runnable() {
152 <                public void run() {
153 <                    try {
154 <                        SynchronousQueue q = new SynchronousQueue();
155 <                        q.put(zero);
156 <                        threadShouldThrow();
157 <                    } catch (InterruptedException ie){
158 <                    }  
159 <                }});
160 <        t.start();
134 <        try {
135 <           Thread.sleep(SHORT_DELAY_MS);
136 <           t.interrupt();
137 <           t.join();
138 <        }
139 <        catch (InterruptedException ie) {
140 <            unexpectedException();
141 <        }
150 >    public void testBlockingPut() throws InterruptedException {
151 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
152 >            public void realRun() throws InterruptedException {
153 >                SynchronousQueue q = new SynchronousQueue();
154 >                q.put(zero);
155 >            }});
156 >
157 >        t.start();
158 >        Thread.sleep(SHORT_DELAY_MS);
159 >        t.interrupt();
160 >        t.join();
161      }
162  
163      /**
164 <     * put blocks waiting for take
164 >     * put blocks waiting for take
165       */
166 <    public void testPutWithTake() {
166 >    public void testPutWithTake() throws InterruptedException {
167          final SynchronousQueue q = new SynchronousQueue();
168 <        Thread t = new Thread(new Runnable() {
169 <                public void run() {
170 <                    int added = 0;
171 <                    try {
172 <                        q.put(new Object());
173 <                        ++added;
155 <                        q.put(new Object());
156 <                        ++added;
157 <                        q.put(new Object());
158 <                        ++added;
159 <                        q.put(new Object());
168 >        Thread t = new Thread(new CheckedRunnable() {
169 >            public void realRun() throws InterruptedException {
170 >                int added = 0;
171 >                try {
172 >                    while (true) {
173 >                        q.put(added);
174                          ++added;
161                        threadShouldThrow();
162                    } catch (InterruptedException e){
163                        assertTrue(added >= 1);
175                      }
176 +                } catch (InterruptedException success) {
177 +                    assertTrue(added == 1);
178                  }
179 <            });
180 <        try {
181 <            t.start();
182 <            Thread.sleep(SHORT_DELAY_MS);
183 <            q.take();
184 <            Thread.sleep(SHORT_DELAY_MS);
185 <            t.interrupt();
186 <            t.join();
174 <        } catch (Exception e){
175 <            unexpectedException();
176 <        }
179 >            }});
180 >
181 >        t.start();
182 >        Thread.sleep(SHORT_DELAY_MS);
183 >        assertEquals(0, q.take());
184 >        Thread.sleep(SHORT_DELAY_MS);
185 >        t.interrupt();
186 >        t.join();
187      }
188  
189      /**
190       * timed offer times out if elements not taken
191       */
192 <    public void testTimedOffer() {
192 >    public void testTimedOffer() throws InterruptedException {
193          final SynchronousQueue q = new SynchronousQueue();
194 <        Thread t = new Thread(new Runnable() {
195 <                public void run() {
196 <                    try {
197 <
198 <                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
199 <                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
200 <                        threadShouldThrow();
201 <                    } catch (InterruptedException success){}
202 <                }
203 <            });
194 <        
195 <        try {
196 <            t.start();
197 <            Thread.sleep(SMALL_DELAY_MS);
198 <            t.interrupt();
199 <            t.join();
200 <        } catch (Exception e){
201 <            unexpectedException();
202 <        }
194 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
195 >            public void realRun() throws InterruptedException {
196 >                assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
197 >                q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
198 >            }});
199 >
200 >        t.start();
201 >        Thread.sleep(SMALL_DELAY_MS);
202 >        t.interrupt();
203 >        t.join();
204      }
205  
206  
207      /**
208       * take blocks interruptibly when empty
209       */
210 <    public void testTakeFromEmpty() {
210 >    public void testTakeFromEmpty() throws InterruptedException {
211          final SynchronousQueue q = new SynchronousQueue();
212 <        Thread t = new Thread(new Runnable() {
213 <                public void run() {
214 <                    try {
215 <                        q.take();
216 <                        threadShouldThrow();
217 <                    } catch (InterruptedException success){ }                
212 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
213 >            public void realRun() throws InterruptedException {
214 >                q.take();
215 >            }});
216 >
217 >        t.start();
218 >        Thread.sleep(SHORT_DELAY_MS);
219 >        t.interrupt();
220 >        t.join();
221 >    }
222 >
223 >
224 >    /**
225 >     * put blocks interruptibly if no active taker
226 >     */
227 >    public void testFairBlockingPut() throws InterruptedException {
228 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
229 >            public void realRun() throws InterruptedException {
230 >                SynchronousQueue q = new SynchronousQueue(true);
231 >                q.put(zero);
232 >            }});
233 >
234 >        t.start();
235 >        Thread.sleep(SHORT_DELAY_MS);
236 >        t.interrupt();
237 >        t.join();
238 >    }
239 >
240 >    /**
241 >     * put blocks waiting for take
242 >     */
243 >    public void testFairPutWithTake() throws InterruptedException {
244 >        final SynchronousQueue q = new SynchronousQueue(true);
245 >        Thread t = new Thread(new CheckedRunnable() {
246 >            public void realRun() throws InterruptedException {
247 >                int added = 0;
248 >                try {
249 >                    while (true) {
250 >                        q.put(added);
251 >                        ++added;
252 >                    }
253 >                } catch (InterruptedException success) {
254 >                    assertTrue(added == 1);
255                  }
256 <            });
257 <        try {
258 <            t.start();
259 <            Thread.sleep(SHORT_DELAY_MS);
260 <            t.interrupt();
261 <            t.join();
262 <        } catch (Exception e){
263 <            unexpectedException();
264 <        }
256 >            }});
257 >
258 >        t.start();
259 >        Thread.sleep(SHORT_DELAY_MS);
260 >        assertEquals(0, q.take());
261 >        Thread.sleep(SHORT_DELAY_MS);
262 >        t.interrupt();
263 >        t.join();
264 >    }
265 >
266 >    /**
267 >     * timed offer times out if elements not taken
268 >     */
269 >    public void testFairTimedOffer() throws InterruptedException {
270 >        final SynchronousQueue q = new SynchronousQueue(true);
271 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
272 >            public void realRun() throws InterruptedException {
273 >                assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
274 >                q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
275 >            }});
276 >
277 >        t.start();
278 >        Thread.sleep(SMALL_DELAY_MS);
279 >        t.interrupt();
280 >        t.join();
281 >    }
282 >
283 >
284 >    /**
285 >     * take blocks interruptibly when empty
286 >     */
287 >    public void testFairTakeFromEmpty() throws InterruptedException {
288 >        final SynchronousQueue q = new SynchronousQueue(true);
289 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
290 >            public void realRun() throws InterruptedException {
291 >                q.take();
292 >            }});
293 >
294 >        t.start();
295 >        Thread.sleep(SHORT_DELAY_MS);
296 >        t.interrupt();
297 >        t.join();
298      }
299  
300      /**
# Line 231 | Line 302 | public class SynchronousQueueTest extend
302       */
303      public void testPoll() {
304          SynchronousQueue q = new SynchronousQueue();
305 <        assertNull(q.poll());
305 >        assertNull(q.poll());
306      }
307  
308      /**
309       * timed pool with zero timeout times out if no active taker
310       */
311 <    public void testTimedPoll0() {
312 <        try {
313 <            SynchronousQueue q = new SynchronousQueue();
243 <            assertNull(q.poll(0, TimeUnit.MILLISECONDS));
244 <        } catch (InterruptedException e){
245 <            unexpectedException();
246 <        }  
311 >    public void testTimedPoll0() throws InterruptedException {
312 >        SynchronousQueue q = new SynchronousQueue();
313 >        assertNull(q.poll(0, MILLISECONDS));
314      }
315  
316      /**
317       * timed pool with nonzero timeout times out if no active taker
318       */
319 <    public void testTimedPoll() {
320 <        try {
321 <            SynchronousQueue q = new SynchronousQueue();
255 <            assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
256 <        } catch (InterruptedException e){
257 <            unexpectedException();
258 <        }  
319 >    public void testTimedPoll() throws InterruptedException {
320 >        SynchronousQueue q = new SynchronousQueue();
321 >        assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
322      }
323  
324      /**
325       * Interrupted timed poll throws InterruptedException instead of
326       * returning timeout status
327       */
328 <    public void testInterruptedTimedPoll() {
329 <        Thread t = new Thread(new Runnable() {
330 <                public void run() {
331 <                    try {
332 <                        SynchronousQueue q = new SynchronousQueue();
333 <                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
334 <                    } catch (InterruptedException success){
335 <                    }  
336 <                }});
337 <        t.start();
338 <        try {
276 <           Thread.sleep(SHORT_DELAY_MS);
277 <           t.interrupt();
278 <           t.join();
279 <        }
280 <        catch (InterruptedException ie) {
281 <            unexpectedException();
282 <        }
328 >    public void testInterruptedTimedPoll() throws InterruptedException {
329 >        final SynchronousQueue q = new SynchronousQueue();
330 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
331 >            public void realRun() throws InterruptedException {
332 >                q.poll(SMALL_DELAY_MS, MILLISECONDS);
333 >            }});
334 >
335 >        t.start();
336 >        Thread.sleep(SHORT_DELAY_MS);
337 >        t.interrupt();
338 >        t.join();
339      }
340  
341      /**
342       *  timed poll before a delayed offer fails; after offer succeeds;
343       *  on interruption throws
344       */
345 <    public void testTimedPollWithOffer() {
345 >    public void testTimedPollWithOffer() throws InterruptedException {
346          final SynchronousQueue q = new SynchronousQueue();
347 <        Thread t = new Thread(new Runnable() {
348 <                public void run() {
349 <                    try {
350 <                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
351 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
352 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
353 <                        threadShouldThrow();
354 <                    } catch (InterruptedException success) { }                
355 <                }
356 <            });
357 <        try {
358 <            t.start();
359 <            Thread.sleep(SMALL_DELAY_MS);
360 <            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
361 <            t.interrupt();
362 <            t.join();
363 <        } catch (Exception e){
364 <            unexpectedException();
365 <        }
366 <    }  
347 >        Thread t = new Thread(new CheckedRunnable() {
348 >            public void realRun() throws InterruptedException {
349 >                assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
350 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
351 >                try {
352 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
353 >                    shouldThrow();
354 >                } catch (InterruptedException success) {}
355 >            }});
356 >
357 >        t.start();
358 >        Thread.sleep(SMALL_DELAY_MS);
359 >        assertTrue(q.offer(zero, SHORT_DELAY_MS, MILLISECONDS));
360 >        t.interrupt();
361 >        t.join();
362 >    }
363 >
364 >    /**
365 >     * Interrupted timed poll throws InterruptedException instead of
366 >     * returning timeout status
367 >     */
368 >    public void testFairInterruptedTimedPoll() throws InterruptedException {
369 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
370 >            public void realRun() throws InterruptedException {
371 >                SynchronousQueue q = new SynchronousQueue(true);
372 >                q.poll(SMALL_DELAY_MS, MILLISECONDS);
373 >            }});
374 >
375 >        t.start();
376 >        Thread.sleep(SHORT_DELAY_MS);
377 >        t.interrupt();
378 >        t.join();
379 >    }
380 >
381 >    /**
382 >     *  timed poll before a delayed offer fails; after offer succeeds;
383 >     *  on interruption throws
384 >     */
385 >    public void testFairTimedPollWithOffer() throws InterruptedException {
386 >        final SynchronousQueue q = new SynchronousQueue(true);
387 >        Thread t = new Thread(new CheckedRunnable() {
388 >            public void realRun() throws InterruptedException {
389 >                assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
390 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
391 >                try {
392 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
393 >                    threadShouldThrow();
394 >                } catch (InterruptedException success) {}
395 >            }});
396 >
397 >        t.start();
398 >        Thread.sleep(SMALL_DELAY_MS);
399 >        assertTrue(q.offer(zero, SHORT_DELAY_MS, MILLISECONDS));
400 >        t.interrupt();
401 >        t.join();
402 >    }
403  
404  
405      /**
# Line 315 | Line 407 | public class SynchronousQueueTest extend
407       */
408      public void testPeek() {
409          SynchronousQueue q = new SynchronousQueue();
410 <        assertNull(q.peek());
410 >        assertNull(q.peek());
411      }
412  
413      /**
# Line 326 | Line 418 | public class SynchronousQueueTest extend
418          try {
419              q.element();
420              shouldThrow();
421 <        }
330 <        catch (NoSuchElementException success) {}
421 >        } catch (NoSuchElementException success) {}
422      }
423  
424      /**
# Line 338 | Line 429 | public class SynchronousQueueTest extend
429          try {
430              q.remove();
431              shouldThrow();
432 <        } catch (NoSuchElementException success){
342 <        }  
432 >        } catch (NoSuchElementException success) {}
433      }
434  
435      /**
# Line 350 | Line 440 | public class SynchronousQueueTest extend
440          assertFalse(q.remove(zero));
441          assertTrue(q.isEmpty());
442      }
443 <        
443 >
444      /**
445       * contains returns false
446       */
# Line 407 | Line 497 | public class SynchronousQueueTest extend
497       */
498      public void testToArray() {
499          SynchronousQueue q = new SynchronousQueue();
500 <        Object[] o = q.toArray();
500 >        Object[] o = q.toArray();
501          assertEquals(o.length, 0);
502      }
503  
# Line 416 | Line 506 | public class SynchronousQueueTest extend
506       */
507      public void testToArray2() {
508          SynchronousQueue q = new SynchronousQueue();
509 <        Integer[] ints = new Integer[1];
509 >        Integer[] ints = new Integer[1];
510          assertNull(ints[0]);
511      }
512 <    
512 >
513 >    /**
514 >     * toArray(null) throws NPE
515 >     */
516 >    public void testToArray_BadArg() {
517 >        SynchronousQueue q = new SynchronousQueue();
518 >        try {
519 >            Object o[] = q.toArray(null);
520 >            shouldThrow();
521 >        } catch (NullPointerException success) {}
522 >    }
523 >
524 >
525      /**
526       * iterator does not traverse any elements
527       */
528      public void testIterator() {
529          SynchronousQueue q = new SynchronousQueue();
530 <        Iterator it = q.iterator();
530 >        Iterator it = q.iterator();
531          assertFalse(it.hasNext());
532          try {
533              Object x = it.next();
534              shouldThrow();
535 <        }
434 <        catch (NoSuchElementException success) {}
535 >        } catch (NoSuchElementException success) {}
536      }
537  
538      /**
# Line 439 | Line 540 | public class SynchronousQueueTest extend
540       */
541      public void testIteratorRemove() {
542          SynchronousQueue q = new SynchronousQueue();
543 <        Iterator it = q.iterator();
543 >        Iterator it = q.iterator();
544          try {
545              it.remove();
546              shouldThrow();
547 <        }
447 <        catch (IllegalStateException success) {}
547 >        } catch (IllegalStateException success) {}
548      }
549  
550      /**
# Line 454 | Line 554 | public class SynchronousQueueTest extend
554          SynchronousQueue q = new SynchronousQueue();
555          String s = q.toString();
556          assertNotNull(s);
557 <    }        
557 >    }
558  
559  
560      /**
# Line 463 | Line 563 | public class SynchronousQueueTest extend
563      public void testOfferInExecutor() {
564          final SynchronousQueue q = new SynchronousQueue();
565          ExecutorService executor = Executors.newFixedThreadPool(2);
466        final Integer one = new Integer(1);
566  
567 <        executor.execute(new Runnable() {
568 <            public void run() {
569 <                threadAssertFalse(q.offer(one));
570 <                try {
571 <                    threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
572 <                    threadAssertEquals(0, q.remainingCapacity());
573 <                }
574 <                catch (InterruptedException e) {
575 <                    threadUnexpectedException();
576 <                }
577 <            }
578 <        });
567 >        executor.execute(new CheckedRunnable() {
568 >            public void realRun() throws InterruptedException {
569 >                assertFalse(q.offer(one));
570 >                assertTrue(q.offer(one, MEDIUM_DELAY_MS, MILLISECONDS));
571 >                assertEquals(0, q.remainingCapacity());
572 >            }});
573 >
574 >        executor.execute(new CheckedRunnable() {
575 >            public void realRun() throws InterruptedException {
576 >                Thread.sleep(SMALL_DELAY_MS);
577 >                assertSame(one, q.take());
578 >            }});
579  
481        executor.execute(new Runnable() {
482            public void run() {
483                try {
484                    Thread.sleep(SMALL_DELAY_MS);
485                    threadAssertEquals(one, q.take());
486                }
487                catch (InterruptedException e) {
488                    threadUnexpectedException();
489                }
490            }
491        });
492        
580          joinPool(executor);
494
581      }
582  
583      /**
# Line 500 | Line 586 | public class SynchronousQueueTest extend
586      public void testPollInExecutor() {
587          final SynchronousQueue q = new SynchronousQueue();
588          ExecutorService executor = Executors.newFixedThreadPool(2);
589 <        executor.execute(new Runnable() {
590 <            public void run() {
591 <                threadAssertNull(q.poll());
592 <                try {
593 <                    threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
594 <                    threadAssertTrue(q.isEmpty());
595 <                }
596 <                catch (InterruptedException e) {
597 <                    threadUnexpectedException();
598 <                }
599 <            }
600 <        });
589 >        executor.execute(new CheckedRunnable() {
590 >            public void realRun() throws InterruptedException {
591 >                assertNull(q.poll());
592 >                assertSame(one, q.poll(MEDIUM_DELAY_MS, MILLISECONDS));
593 >                assertTrue(q.isEmpty());
594 >            }});
595 >
596 >        executor.execute(new CheckedRunnable() {
597 >            public void realRun() throws InterruptedException {
598 >                Thread.sleep(SHORT_DELAY_MS);
599 >                q.put(one);
600 >            }});
601  
516        executor.execute(new Runnable() {
517            public void run() {
518                try {
519                    Thread.sleep(SMALL_DELAY_MS);
520                    q.put(new Integer(1));
521                }
522                catch (InterruptedException e) {
523                    threadUnexpectedException();
524                }
525            }
526        });
527        
602          joinPool(executor);
603      }
604  
605      /**
606       * a deserialized serialized queue is usable
607       */
608 <    public void testSerialization() {
608 >    public void testSerialization() throws Exception {
609 >        SynchronousQueue q = new SynchronousQueue();
610 >        ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
611 >        ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
612 >        out.writeObject(q);
613 >        out.close();
614 >
615 >        ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
616 >        ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
617 >        SynchronousQueue r = (SynchronousQueue)in.readObject();
618 >        assertEquals(q.size(), r.size());
619 >        while (!q.isEmpty())
620 >            assertEquals(q.remove(), r.remove());
621 >    }
622 >
623 >    /**
624 >     * drainTo(null) throws NPE
625 >     */
626 >    public void testDrainToNull() {
627 >        SynchronousQueue q = new SynchronousQueue();
628 >        try {
629 >            q.drainTo(null);
630 >            shouldThrow();
631 >        } catch (NullPointerException success) {}
632 >    }
633 >
634 >    /**
635 >     * drainTo(this) throws IAE
636 >     */
637 >    public void testDrainToSelf() {
638          SynchronousQueue q = new SynchronousQueue();
639          try {
640 <            ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
641 <            ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
642 <            out.writeObject(q);
643 <            out.close();
644 <
645 <            ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
646 <            ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
647 <            SynchronousQueue r = (SynchronousQueue)in.readObject();
648 <            assertEquals(q.size(), r.size());
649 <            while (!q.isEmpty())
650 <                assertEquals(q.remove(), r.remove());
651 <        } catch(Exception e){
652 <            unexpectedException();
653 <        }
640 >            q.drainTo(q);
641 >            shouldThrow();
642 >        } catch (IllegalArgumentException success) {}
643 >    }
644 >
645 >    /**
646 >     * drainTo(c) of empty queue doesn't transfer elements
647 >     */
648 >    public void testDrainTo() {
649 >        SynchronousQueue q = new SynchronousQueue();
650 >        ArrayList l = new ArrayList();
651 >        q.drainTo(l);
652 >        assertEquals(q.size(), 0);
653 >        assertEquals(l.size(), 0);
654 >    }
655 >
656 >    /**
657 >     * drainTo empties queue, unblocking a waiting put.
658 >     */
659 >    public void testDrainToWithActivePut() throws InterruptedException {
660 >        final SynchronousQueue q = new SynchronousQueue();
661 >        Thread t = new Thread(new CheckedRunnable() {
662 >            public void realRun() throws InterruptedException {
663 >                q.put(new Integer(1));
664 >            }});
665 >
666 >        t.start();
667 >        ArrayList l = new ArrayList();
668 >        Thread.sleep(SHORT_DELAY_MS);
669 >        q.drainTo(l);
670 >        assertTrue(l.size() <= 1);
671 >        if (l.size() > 0)
672 >            assertEquals(l.get(0), new Integer(1));
673 >        t.join();
674 >        assertTrue(l.size() <= 1);
675 >    }
676 >
677 >    /**
678 >     * drainTo(null, n) throws NPE
679 >     */
680 >    public void testDrainToNullN() {
681 >        SynchronousQueue q = new SynchronousQueue();
682 >        try {
683 >            q.drainTo(null, 0);
684 >            shouldThrow();
685 >        } catch (NullPointerException success) {}
686 >    }
687 >
688 >    /**
689 >     * drainTo(this, n) throws IAE
690 >     */
691 >    public void testDrainToSelfN() {
692 >        SynchronousQueue q = new SynchronousQueue();
693 >        try {
694 >            q.drainTo(q, 0);
695 >            shouldThrow();
696 >        } catch (IllegalArgumentException success) {}
697 >    }
698 >
699 >    /**
700 >     * drainTo(c, n) empties up to n elements of queue into c
701 >     */
702 >    public void testDrainToN() throws InterruptedException {
703 >        final SynchronousQueue q = new SynchronousQueue();
704 >        Thread t1 = new Thread(new CheckedRunnable() {
705 >            public void realRun() throws InterruptedException {
706 >                q.put(one);
707 >            }});
708 >
709 >        Thread t2 = new Thread(new CheckedRunnable() {
710 >            public void realRun() throws InterruptedException {
711 >                q.put(two);
712 >            }});
713 >
714 >        t1.start();
715 >        t2.start();
716 >        ArrayList l = new ArrayList();
717 >        Thread.sleep(SHORT_DELAY_MS);
718 >        q.drainTo(l, 1);
719 >        assertTrue(l.size() == 1);
720 >        q.drainTo(l, 1);
721 >        assertTrue(l.size() == 2);
722 >        assertTrue(l.contains(one));
723 >        assertTrue(l.contains(two));
724 >        t1.join();
725 >        t2.join();
726      }
727  
728   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines