ConsumableBlockingQueue extends ArrayBlockingQueue<T> { @Override @Nonnull public Iterator<T> iterator() { return new Iterator<T>() { @Override public boolean hasNext() { return !isEmpty(); } @Override public T next() { if (!hasNext()) { throw new NoSuchElementException("No more items"); } return poll(); } }; } ConsumableBlockingQueue(final int maxSize); @Override @Nonnull Iterator<T> iterator(); @Override boolean equals(final Object obj); @Override int hashCode(); @Override String toString();  }
@Test public void shouldBlockOnAdditionWhenQueueIsFull() throws InterruptedException { final ConsumableBlockingQueue<Integer> queue = new ConsumableBlockingQueue<>(5); final boolean[] finishedAdding = new boolean[] {false}; new Thread(() -> { IntStream.range(0, 10) .forEach(i -> { try { queue.put(i); } catch (InterruptedException e) { throw new RuntimeException(e); } }); finishedAdding[0] = true; }).start(); Thread.sleep(1000L); assertFalse(finishedAdding[0]); final Iterator<Integer> consumer = queue.iterator(); final List<Integer> items = new ArrayList<>(); for (int i = 0; i < 4; i++) { assertTrue(consumer.hasNext()); items.add(consumer.next()); } Thread.sleep(1000L); assertFalse(finishedAdding[0]); for (int i = 0; i < 4; i++) { assertTrue(consumer.hasNext()); items.add(consumer.next()); } Thread.sleep(1000L); assertTrue(finishedAdding[0]); while (consumer.hasNext()) { items.add(consumer.next()); } assertEquals(Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9), items); }
@Test public void shouldNotBlockWhenConsumingWhenQueueIsEmpty() { final ConsumableBlockingQueue<Integer> queue = new ConsumableBlockingQueue<>(5); final Iterator<Integer> iterator = queue.iterator(); assertFalse(iterator.hasNext()); }