BlockedForEachLinkedBlockingQueue.java

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class BlockedForEachLinkedBlockingQueue<T> extends LinkedBlockingQueue<T> {

	private static final long serialVersionUID = 9053130343708158526L;
	private static final AtomicBoolean stopBlockedForEach = new AtomicBoolean(false);

	public void stopBlockedForEach() {
		stopBlockedForEach.set(true);
	}

	public void blockedForEachWrappingException(Consumer<? super T> action) {
		Consumer<T> consumer = t -> action.accept(t);
		Supplier<T> supplier = () -> takeWrappingToUncatchedException();
		while (!stopBlockedForEach.get()) {
			consumer.accept(supplier.get());
		}
	}

	private T takeWrappingToUncatchedException() {
		try {
			return take();
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}
}