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.13 by jsr166, Thu Nov 19 01:15:42 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() {
# Line 32 | Line 34 | public class SynchronousQueueTest extend
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 +    }
46 +
47 +    /**
48       * offer(null) throws NPE
49       */
50      public void testOfferNull() {
# Line 39 | Line 52 | public class SynchronousQueueTest extend
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 59 | Line 83 | public class SynchronousQueueTest extend
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 <        }
75 <        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() {
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;
174 <                        q.put(new Object());
175 <                        ++added;
176 <                        q.put(new Object());
177 <                        ++added;
178 <                        q.put(new Object());
179 <                        ++added;
180 <                        threadShouldThrow();
181 <                    } catch (InterruptedException e){
182 <                        assertTrue(added >= 1);
164 <                    }
168 >        Thread t = new Thread(new CheckedRunnable() {
169 >            public void realRun() throws InterruptedException {
170 >                int added = 0;
171 >                try {
172 >                    q.put(new Object());
173 >                    ++added;
174 >                    q.put(new Object());
175 >                    ++added;
176 >                    q.put(new Object());
177 >                    ++added;
178 >                    q.put(new Object());
179 >                    ++added;
180 >                    threadShouldThrow();
181 >                } catch (InterruptedException success) {
182 >                    assertTrue(added >= 1);
183                  }
184 <            });
185 <        try {
186 <            t.start();
187 <            Thread.sleep(SHORT_DELAY_MS);
188 <            q.take();
189 <            Thread.sleep(SHORT_DELAY_MS);
190 <            t.interrupt();
191 <            t.join();
174 <        } catch (Exception e){
175 <            unexpectedException();
176 <        }
184 >            }});
185 >
186 >        t.start();
187 >        Thread.sleep(SHORT_DELAY_MS);
188 >        q.take();
189 >        Thread.sleep(SHORT_DELAY_MS);
190 >        t.interrupt();
191 >        t.join();
192      }
193  
194      /**
195       * timed offer times out if elements not taken
196       */
197 <    public void testTimedOffer() {
197 >    public void testTimedOffer() throws InterruptedException {
198          final SynchronousQueue q = new SynchronousQueue();
199 <        Thread t = new Thread(new Runnable() {
200 <                public void run() {
201 <                    try {
202 <
203 <                        threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
204 <                        q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS);
205 <                        threadShouldThrow();
206 <                    } catch (InterruptedException success){}
207 <                }
208 <            });
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 <        }
199 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
200 >            public void realRun() throws InterruptedException {
201 >                threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
202 >                q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
203 >            }});
204 >
205 >        t.start();
206 >        Thread.sleep(SMALL_DELAY_MS);
207 >        t.interrupt();
208 >        t.join();
209      }
210  
211  
212      /**
213       * take blocks interruptibly when empty
214       */
215 <    public void testTakeFromEmpty() {
215 >    public void testTakeFromEmpty() throws InterruptedException {
216          final SynchronousQueue q = new SynchronousQueue();
217 <        Thread t = new Thread(new Runnable() {
218 <                public void run() {
219 <                    try {
220 <                        q.take();
221 <                        threadShouldThrow();
222 <                    } catch (InterruptedException success){ }                
217 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
218 >            public void realRun() throws InterruptedException {
219 >                q.take();
220 >            }});
221 >
222 >        t.start();
223 >        Thread.sleep(SHORT_DELAY_MS);
224 >        t.interrupt();
225 >        t.join();
226 >    }
227 >
228 >
229 >    /**
230 >     * put blocks interruptibly if no active taker
231 >     */
232 >    public void testFairBlockingPut() throws InterruptedException {
233 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
234 >            public void realRun() throws InterruptedException {
235 >                SynchronousQueue q = new SynchronousQueue(true);
236 >                q.put(zero);
237 >            }});
238 >
239 >        t.start();
240 >        Thread.sleep(SHORT_DELAY_MS);
241 >        t.interrupt();
242 >        t.join();
243 >    }
244 >
245 >    /**
246 >     * put blocks waiting for take
247 >     */
248 >    public void testFairPutWithTake() throws InterruptedException {
249 >        final SynchronousQueue q = new SynchronousQueue(true);
250 >        Thread t = new Thread(new CheckedRunnable() {
251 >            public void realRun() throws InterruptedException {
252 >                int added = 0;
253 >                try {
254 >                    q.put(new Object());
255 >                    ++added;
256 >                    q.put(new Object());
257 >                    ++added;
258 >                    q.put(new Object());
259 >                    ++added;
260 >                    q.put(new Object());
261 >                    ++added;
262 >                    threadShouldThrow();
263 >                } catch (InterruptedException success) {
264 >                    assertTrue(added >= 1);
265                  }
266 <            });
267 <        try {
268 <            t.start();
269 <            Thread.sleep(SHORT_DELAY_MS);
270 <            t.interrupt();
271 <            t.join();
272 <        } catch (Exception e){
273 <            unexpectedException();
274 <        }
266 >            }});
267 >
268 >        t.start();
269 >        Thread.sleep(SHORT_DELAY_MS);
270 >        q.take();
271 >        Thread.sleep(SHORT_DELAY_MS);
272 >        t.interrupt();
273 >        t.join();
274 >    }
275 >
276 >    /**
277 >     * timed offer times out if elements not taken
278 >     */
279 >    public void testFairTimedOffer() throws InterruptedException {
280 >        final SynchronousQueue q = new SynchronousQueue(true);
281 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
282 >            public void realRun() throws InterruptedException {
283 >                threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS));
284 >                q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS);
285 >            }});
286 >
287 >        t.start();
288 >        Thread.sleep(SMALL_DELAY_MS);
289 >        t.interrupt();
290 >        t.join();
291 >    }
292 >
293 >
294 >    /**
295 >     * take blocks interruptibly when empty
296 >     */
297 >    public void testFairTakeFromEmpty() throws InterruptedException {
298 >        final SynchronousQueue q = new SynchronousQueue(true);
299 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
300 >            public void realRun() throws InterruptedException {
301 >                q.take();
302 >            }});
303 >
304 >        t.start();
305 >        Thread.sleep(SHORT_DELAY_MS);
306 >        t.interrupt();
307 >        t.join();
308      }
309  
310      /**
# Line 237 | Line 318 | public class SynchronousQueueTest extend
318      /**
319       * timed pool with zero timeout times out if no active taker
320       */
321 <    public void testTimedPoll0() {
322 <        try {
323 <            SynchronousQueue q = new SynchronousQueue();
243 <            assertNull(q.poll(0, TimeUnit.MILLISECONDS));
244 <        } catch (InterruptedException e){
245 <            unexpectedException();
246 <        }  
321 >    public void testTimedPoll0() throws InterruptedException {
322 >        SynchronousQueue q = new SynchronousQueue();
323 >        assertNull(q.poll(0, MILLISECONDS));
324      }
325  
326      /**
327       * timed pool with nonzero timeout times out if no active taker
328       */
329 <    public void testTimedPoll() {
330 <        try {
331 <            SynchronousQueue q = new SynchronousQueue();
255 <            assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
256 <        } catch (InterruptedException e){
257 <            unexpectedException();
258 <        }  
329 >    public void testTimedPoll() throws InterruptedException {
330 >        SynchronousQueue q = new SynchronousQueue();
331 >        assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
332      }
333  
334      /**
335       * Interrupted timed poll throws InterruptedException instead of
336       * returning timeout status
337       */
338 <    public void testInterruptedTimedPoll() {
339 <        Thread t = new Thread(new Runnable() {
340 <                public void run() {
341 <                    try {
342 <                        SynchronousQueue q = new SynchronousQueue();
343 <                        assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
344 <                    } catch (InterruptedException success){
345 <                    }  
346 <                }});
347 <        t.start();
348 <        try {
276 <           Thread.sleep(SHORT_DELAY_MS);
277 <           t.interrupt();
278 <           t.join();
279 <        }
280 <        catch (InterruptedException ie) {
281 <            unexpectedException();
282 <        }
338 >    public void testInterruptedTimedPoll() throws InterruptedException {
339 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
340 >            public void realRun() throws InterruptedException {
341 >                SynchronousQueue q = new SynchronousQueue();
342 >                q.poll(SMALL_DELAY_MS, MILLISECONDS);
343 >            }});
344 >
345 >        t.start();
346 >        Thread.sleep(SHORT_DELAY_MS);
347 >        t.interrupt();
348 >        t.join();
349      }
350  
351      /**
352       *  timed poll before a delayed offer fails; after offer succeeds;
353       *  on interruption throws
354       */
355 <    public void testTimedPollWithOffer() {
355 >    public void testTimedPollWithOffer() throws InterruptedException {
356          final SynchronousQueue q = new SynchronousQueue();
357 <        Thread t = new Thread(new Runnable() {
358 <                public void run() {
359 <                    try {
360 <                        threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
361 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
362 <                        q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS);
363 <                        threadShouldThrow();
364 <                    } catch (InterruptedException success) { }                
365 <                }
366 <            });
367 <        try {
368 <            t.start();
369 <            Thread.sleep(SMALL_DELAY_MS);
370 <            assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS));
371 <            t.interrupt();
372 <            t.join();
373 <        } catch (Exception e){
374 <            unexpectedException();
375 <        }
376 <    }  
357 >        Thread t = new Thread(new CheckedRunnable() {
358 >            public void realRun() throws InterruptedException {
359 >                threadAssertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
360 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
361 >                try {
362 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
363 >                    threadShouldThrow();
364 >                } catch (InterruptedException success) {}
365 >            }});
366 >
367 >        t.start();
368 >        Thread.sleep(SMALL_DELAY_MS);
369 >        assertTrue(q.offer(zero, SHORT_DELAY_MS, MILLISECONDS));
370 >        t.interrupt();
371 >        t.join();
372 >    }
373 >
374 >    /**
375 >     * Interrupted timed poll throws InterruptedException instead of
376 >     * returning timeout status
377 >     */
378 >    public void testFairInterruptedTimedPoll() throws InterruptedException {
379 >        Thread t = new Thread(new CheckedInterruptedRunnable() {
380 >            public void realRun() throws InterruptedException {
381 >                SynchronousQueue q = new SynchronousQueue(true);
382 >                q.poll(SMALL_DELAY_MS, MILLISECONDS);
383 >            }});
384 >
385 >        t.start();
386 >        Thread.sleep(SHORT_DELAY_MS);
387 >        t.interrupt();
388 >        t.join();
389 >    }
390 >
391 >    /**
392 >     *  timed poll before a delayed offer fails; after offer succeeds;
393 >     *  on interruption throws
394 >     */
395 >    public void testFairTimedPollWithOffer() throws InterruptedException {
396 >        final SynchronousQueue q = new SynchronousQueue(true);
397 >        Thread t = new Thread(new CheckedRunnable() {
398 >            public void realRun() throws InterruptedException {
399 >                threadAssertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS));
400 >                assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS));
401 >                try {
402 >                    q.poll(LONG_DELAY_MS, MILLISECONDS);
403 >                    threadShouldThrow();
404 >                } catch (InterruptedException success) {}
405 >            }});
406 >
407 >        t.start();
408 >        Thread.sleep(SMALL_DELAY_MS);
409 >        assertTrue(q.offer(zero, SHORT_DELAY_MS, MILLISECONDS));
410 >        t.interrupt();
411 >        t.join();
412 >    }
413  
414  
415      /**
# Line 326 | Line 428 | public class SynchronousQueueTest extend
428          try {
429              q.element();
430              shouldThrow();
431 <        }
330 <        catch (NoSuchElementException success) {}
431 >        } catch (NoSuchElementException success) {}
432      }
433  
434      /**
# Line 338 | Line 439 | public class SynchronousQueueTest extend
439          try {
440              q.remove();
441              shouldThrow();
442 <        } catch (NoSuchElementException success){
443 <        }  
442 >        } catch (NoSuchElementException success) {
443 >        }
444      }
445  
446      /**
# Line 350 | Line 451 | public class SynchronousQueueTest extend
451          assertFalse(q.remove(zero));
452          assertTrue(q.isEmpty());
453      }
454 <        
454 >
455      /**
456       * contains returns false
457       */
# Line 419 | Line 520 | public class SynchronousQueueTest extend
520          Integer[] ints = new Integer[1];
521          assertNull(ints[0]);
522      }
523 <    
523 >
524 >    /**
525 >     * toArray(null) throws NPE
526 >     */
527 >    public void testToArray_BadArg() {
528 >        try {
529 >            SynchronousQueue q = new SynchronousQueue();
530 >            Object o[] = q.toArray(null);
531 >            shouldThrow();
532 >        } catch (NullPointerException success) {}
533 >    }
534 >
535 >
536      /**
537       * iterator does not traverse any elements
538       */
# Line 430 | Line 543 | public class SynchronousQueueTest extend
543          try {
544              Object x = it.next();
545              shouldThrow();
546 <        }
434 <        catch (NoSuchElementException success) {}
546 >        } catch (NoSuchElementException success) {}
547      }
548  
549      /**
# Line 443 | Line 555 | public class SynchronousQueueTest extend
555          try {
556              it.remove();
557              shouldThrow();
558 <        }
447 <        catch (IllegalStateException success) {}
558 >        } catch (IllegalStateException success) {}
559      }
560  
561      /**
# Line 454 | Line 565 | public class SynchronousQueueTest extend
565          SynchronousQueue q = new SynchronousQueue();
566          String s = q.toString();
567          assertNotNull(s);
568 <    }        
568 >    }
569  
570  
571      /**
# Line 465 | Line 576 | public class SynchronousQueueTest extend
576          ExecutorService executor = Executors.newFixedThreadPool(2);
577          final Integer one = new Integer(1);
578  
579 <        executor.execute(new Runnable() {
580 <            public void run() {
579 >        executor.execute(new CheckedRunnable() {
580 >            public void realRun() throws InterruptedException {
581                  threadAssertFalse(q.offer(one));
582 <                try {
583 <                    threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
584 <                    threadAssertEquals(0, q.remainingCapacity());
585 <                }
586 <                catch (InterruptedException e) {
587 <                    threadUnexpectedException();
588 <                }
589 <            }
590 <        });
582 >                threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, MILLISECONDS));
583 >                threadAssertEquals(0, q.remainingCapacity());
584 >            }});
585 >
586 >        executor.execute(new CheckedRunnable() {
587 >            public void realRun() throws InterruptedException {
588 >                Thread.sleep(SMALL_DELAY_MS);
589 >                threadAssertEquals(one, q.take());
590 >            }});
591  
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        
592          joinPool(executor);
593  
594      }
# Line 500 | Line 599 | public class SynchronousQueueTest extend
599      public void testPollInExecutor() {
600          final SynchronousQueue q = new SynchronousQueue();
601          ExecutorService executor = Executors.newFixedThreadPool(2);
602 <        executor.execute(new Runnable() {
603 <            public void run() {
602 >        executor.execute(new CheckedRunnable() {
603 >            public void realRun() throws InterruptedException {
604                  threadAssertNull(q.poll());
605 <                try {
606 <                    threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS));
607 <                    threadAssertTrue(q.isEmpty());
608 <                }
609 <                catch (InterruptedException e) {
610 <                    threadUnexpectedException();
611 <                }
612 <            }
613 <        });
605 >                threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, MILLISECONDS));
606 >                threadAssertTrue(q.isEmpty());
607 >            }});
608 >
609 >        executor.execute(new CheckedRunnable() {
610 >            public void realRun() throws InterruptedException {
611 >                Thread.sleep(SMALL_DELAY_MS);
612 >                q.put(new Integer(1));
613 >            }});
614  
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        
615          joinPool(executor);
616      }
617  
618      /**
619       * a deserialized serialized queue is usable
620       */
621 <    public void testSerialization() {
621 >    public void testSerialization() throws Exception {
622 >        SynchronousQueue q = new SynchronousQueue();
623 >        ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
624 >        ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
625 >        out.writeObject(q);
626 >        out.close();
627 >
628 >        ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
629 >        ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
630 >        SynchronousQueue r = (SynchronousQueue)in.readObject();
631 >        assertEquals(q.size(), r.size());
632 >        while (!q.isEmpty())
633 >            assertEquals(q.remove(), r.remove());
634 >    }
635 >
636 >    /**
637 >     * drainTo(null) throws NPE
638 >     */
639 >    public void testDrainToNull() {
640 >        SynchronousQueue q = new SynchronousQueue();
641 >        try {
642 >            q.drainTo(null);
643 >            shouldThrow();
644 >        } catch (NullPointerException success) {}
645 >    }
646 >
647 >    /**
648 >     * drainTo(this) throws IAE
649 >     */
650 >    public void testDrainToSelf() {
651 >        SynchronousQueue q = new SynchronousQueue();
652 >        try {
653 >            q.drainTo(q);
654 >            shouldThrow();
655 >        } catch (IllegalArgumentException success) {}
656 >    }
657 >
658 >    /**
659 >     * drainTo(c) of empty queue doesn't transfer elements
660 >     */
661 >    public void testDrainTo() {
662 >        SynchronousQueue q = new SynchronousQueue();
663 >        ArrayList l = new ArrayList();
664 >        q.drainTo(l);
665 >        assertEquals(q.size(), 0);
666 >        assertEquals(l.size(), 0);
667 >    }
668 >
669 >    /**
670 >     * drainTo empties queue, unblocking a waiting put.
671 >     */
672 >    public void testDrainToWithActivePut() throws InterruptedException {
673 >        final SynchronousQueue q = new SynchronousQueue();
674 >        Thread t = new Thread(new CheckedRunnable() {
675 >            public void realRun() throws InterruptedException {
676 >                q.put(new Integer(1));
677 >            }});
678 >
679 >        t.start();
680 >        ArrayList l = new ArrayList();
681 >        Thread.sleep(SHORT_DELAY_MS);
682 >        q.drainTo(l);
683 >        assertTrue(l.size() <= 1);
684 >        if (l.size() > 0)
685 >            assertEquals(l.get(0), new Integer(1));
686 >        t.join();
687 >        assertTrue(l.size() <= 1);
688 >    }
689 >
690 >    /**
691 >     * drainTo(null, n) throws NPE
692 >     */
693 >    public void testDrainToNullN() {
694 >        SynchronousQueue q = new SynchronousQueue();
695 >        try {
696 >            q.drainTo(null, 0);
697 >            shouldThrow();
698 >        } catch (NullPointerException success) {}
699 >    }
700 >
701 >    /**
702 >     * drainTo(this, n) throws IAE
703 >     */
704 >    public void testDrainToSelfN() {
705          SynchronousQueue q = new SynchronousQueue();
706          try {
707 <            ByteArrayOutputStream bout = new ByteArrayOutputStream(10000);
708 <            ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout));
709 <            out.writeObject(q);
540 <            out.close();
541 <
542 <            ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray());
543 <            ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin));
544 <            SynchronousQueue r = (SynchronousQueue)in.readObject();
545 <            assertEquals(q.size(), r.size());
546 <            while (!q.isEmpty())
547 <                assertEquals(q.remove(), r.remove());
548 <        } catch(Exception e){
549 <            unexpectedException();
707 >            q.drainTo(q, 0);
708 >            shouldThrow();
709 >        } catch (IllegalArgumentException success) {
710          }
711      }
712  
713 +    /**
714 +     * drainTo(c, n) empties up to n elements of queue into c
715 +     */
716 +    public void testDrainToN() throws InterruptedException {
717 +        final SynchronousQueue q = new SynchronousQueue();
718 +        Thread t1 = new Thread(new CheckedRunnable() {
719 +            public void realRun() throws InterruptedException {
720 +                q.put(one);
721 +            }});
722 +
723 +        Thread t2 = new Thread(new CheckedRunnable() {
724 +            public void realRun() throws InterruptedException {
725 +                q.put(two);
726 +            }});
727 +
728 +        t1.start();
729 +        t2.start();
730 +        ArrayList l = new ArrayList();
731 +        Thread.sleep(SHORT_DELAY_MS);
732 +        q.drainTo(l, 1);
733 +        assertTrue(l.size() == 1);
734 +        q.drainTo(l, 1);
735 +        assertTrue(l.size() == 2);
736 +        assertTrue(l.contains(one));
737 +        assertTrue(l.contains(two));
738 +        t1.join();
739 +        t2.join();
740 +    }
741 +
742   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines