@Test public void testBlockingDrain() throws InterruptedException { final BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>(); for (int i = 0; i < Random.getScaleCount(); ++i) { queue.add(Random.getObject()); } queue.drainTo(Lists.newArrayList()); List<Object> buffer = Lists.newArrayList(); final AtomicInteger expected = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @Override public void run() { List<Object> objects = Lists.newArrayList(); for (int i = 0; i < Random.getScaleCount(); ++i) { objects.add(Random.getObject()); } expected.set(objects.size()); queue.addAll(objects); latch.countDown(); } }); Random.sleep(); t.start(); latch.await(); int actual = Queues.blockingDrain(queue, buffer); Assert.assertEquals(expected.get(), actual); Assert.assertNull(queue.poll()); }
public static <E> int blockingDrain(BlockingQueue<E> queue, Collection<? super E> buffer) throws InterruptedException { Preconditions.checkNotNull(buffer); int added = queue.drainTo(buffer); if(added == 0) { buffer.add(queue.take()); added += queue.drainTo(buffer); ++added; } return added; }
Queues { public static <E> int blockingDrain(BlockingQueue<E> queue, Collection<? super E> buffer) throws InterruptedException { Preconditions.checkNotNull(buffer); int added = queue.drainTo(buffer); if(added == 0) { buffer.add(queue.take()); added += queue.drainTo(buffer); ++added; } return added; } }
Queues { public static <E> int blockingDrain(BlockingQueue<E> queue, Collection<? super E> buffer) throws InterruptedException { Preconditions.checkNotNull(buffer); int added = queue.drainTo(buffer); if(added == 0) { buffer.add(queue.take()); added += queue.drainTo(buffer); ++added; } return added; } private  Queues(); }
Queues { public static <E> int blockingDrain(BlockingQueue<E> queue, Collection<? super E> buffer) throws InterruptedException { Preconditions.checkNotNull(buffer); int added = queue.drainTo(buffer); if(added == 0) { buffer.add(queue.take()); added += queue.drainTo(buffer); ++added; } return added; } private  Queues(); static Queue<T> newSingleThreadedQueue(); static int blockingDrain(BlockingQueue<E> queue,
            Collection<? super E> buffer); static int blockingDrain(BlockingQueue<E> queue,
            Collection<? super E> buffer, long timeout, TimeUnit unit); }
Queues { public static <E> int blockingDrain(BlockingQueue<E> queue, Collection<? super E> buffer) throws InterruptedException { Preconditions.checkNotNull(buffer); int added = queue.drainTo(buffer); if(added == 0) { buffer.add(queue.take()); added += queue.drainTo(buffer); ++added; } return added; } private  Queues(); static Queue<T> newSingleThreadedQueue(); static int blockingDrain(BlockingQueue<E> queue,
            Collection<? super E> buffer); static int blockingDrain(BlockingQueue<E> queue,
            Collection<? super E> buffer, long timeout, TimeUnit unit);  }