/** * * Copyright 2014 Florian Schmaus * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package org.jivesoftware.smack.util; import java.util.AbstractQueue; import java.util.Collection; import java.util.Iterator; import java.util.NoSuchElementException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; /** * Like ArrayBlockingQueue but with additional {@link #shutdown()} and {@link #start} methods. Will * throw {@link InterruptedException} if Queue has been shutdown on {@link #take()} and * {@link #poll(long, TimeUnit)}. *
* Based on ArrayBlockingQueue of OpenJDK by Doug Lea (who released ArrayBlockingQueue as public
* domain).
*
* @param
* This may throw an {@link InterruptedException} in two cases
*
*
* So you have to check which is the case, e.g. by calling {@link #isShutdown()}.
*
* @param e the element to add.
* @throws InterruptedException if interrupted while waiting or if the queue was shut down.
*/
public void put(E e) throws InterruptedException {
checkNotNull(e);
lock.lockInterruptibly();
try {
while (isFull()) {
try {
notFull.await();
checkNotShutdown();
}
catch (InterruptedException ie) {
notFull.signal();
throw ie;
}
}
insert(e);
}
finally {
lock.unlock();
}
}
@Override
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
checkNotNull(e);
long nanos = unit.toNanos(timeout);
lock.lockInterruptibly();
try {
while (true) {
if (isNotFull()) {
insert(e);
return true;
}
if (nanos <= 0) {
return false;
}
try {
nanos = notFull.awaitNanos(nanos);
checkNotShutdown();
}
catch (InterruptedException ie) {
notFull.signal();
throw ie;
}
}
}
finally {
lock.unlock();
}
}
@Override
public E take() throws InterruptedException {
lock.lockInterruptibly();
try {
checkNotShutdown();
try {
while (hasNoElements()) {
notEmpty.await();
checkNotShutdown();
}
}
catch (InterruptedException ie) {
notEmpty.signal();
throw ie;
}
E e = extract();
return e;
}
finally {
lock.unlock();
}
}
@Override
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
lock.lockInterruptibly();
try {
checkNotShutdown();
while (true) {
if (hasElements()) {
E e = extract();
return e;
}
if (nanos <= 0) {
return null;
}
try {
nanos = notEmpty.awaitNanos(nanos);
checkNotShutdown();
}
catch (InterruptedException ie) {
notEmpty.signal();
throw ie;
}
}
}
finally {
lock.unlock();
}
}
@Override
public int remainingCapacity() {
lock.lock();
try {
return items.length - count;
}
finally {
lock.unlock();
}
}
@Override
public int drainTo(Collection super E> c) {
checkNotNull(c);
if (c == this) {
throw new IllegalArgumentException();
}
lock.lock();
try {
int i = takeIndex;
int n = 0;
for (; n < count; n++) {
c.add(items[i]);
items[i] = null;
i = inc(i);
}
if (n > 0) {
count = 0;
putIndex = 0;
takeIndex = 0;
notFull.signalAll();
}
return n;
}
finally {
lock.unlock();
}
}
@Override
public int drainTo(Collection super E> c, int maxElements) {
checkNotNull(c);
if (c == this) {
throw new IllegalArgumentException();
}
if (maxElements <= 0) {
return 0;
}
lock.lock();
try {
int i = takeIndex;
int n = 0;
int max = (maxElements < count) ? maxElements : count;
for (; n < max; n++) {
c.add(items[i]);
items[i] = null;
i = inc(i);
}
if (n > 0) {
count -= n;
takeIndex = i;
notFull.signalAll();
}
return n;
}
finally {
lock.unlock();
}
}
@Override
public int size() {
lock.lock();
try {
return count;
}
finally {
lock.unlock();
}
}
@Override
public Iterator