### Question:
SinkQueue { void consumeAll(Consumer<T> consumer) throws InterruptedException { waitForData(); try { for (int i = size(); i-- > 0; ) { consumer.consume(front()); _dequeue(); } } finally { clearConsumerLock(); } } @SuppressWarnings("unchecked")  SinkQueue(int capacity);   }### Answer:
@Test public void testConsumeAll() throws Exception { final int capacity = 64; final SinkQueue<Integer> q = new SinkQueue<Integer>(capacity); for (int i = 0; i < capacity; ++i) { assertTrue("should enqueue", q.enqueue(i)); } assertTrue("should not enqueue", !q.enqueue(capacity)); final Runnable trigger = mock(Runnable.class); q.consumeAll(new Consumer<Integer>() { private int expected = 0; @Override public void consume(Integer e) { assertEquals("element", expected++, (int) e); trigger.run(); } }); verify(trigger, times(capacity)).run(); }