ThresholdBlockingQueue extends AbstractQueue<E> implements BlockingQueue<E> { public synchronized void waitUntilQueueSizeLowerThanThreshold(DateTime waitUntil) throws InterruptedException { while (queue.size() > notifyThreshHold) { long sleep = waitUntil.getMillis() - currentTimeMillis(); if (sleep <= 0) { break; } wait(sleep); } } ThresholdBlockingQueue(int capacity, int notifyThreshHold); synchronized void notifyIfNotFull(); synchronized void waitUntilQueueSizeLowerThanThreshold(DateTime waitUntil); @Override boolean offer(E e); @Override E poll(); @Override E peek(); @Override Iterator<E> iterator(); @Override int size(); @Override void put(E e); @Override boolean offer(E e, long timeout, TimeUnit unit); @Override E take(); @Override E poll(long timeout, TimeUnit unit); @Override int remainingCapacity(); @Override int drainTo(Collection<? super E> c); @Override int drainTo(Collection<? super E> c, int maxElements);  }
@Test public void doesNotWaitIfQueueIsAlreadyBelowThreshold() { assertTimeout(ofMillis(5000), () -> q.waitUntilQueueSizeLowerThanThreshold(DateTime.now().plusMinutes(1))); }