package org.icij.extract.core;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

import java.util.logging.Logger;

import java.nio.file.Path;

/**
 * An implementation of {@link Scanner} which pushes encountered file paths into a
 * given queue. This is a classic producer, putting elements into a queue which
 * are then extracted by a consumer.
 *
 * Paths are pushed into the queue synchronously and if the queue is bounded, only
 * when a space becomes available.
 *
 * A buffer size may be specified in the constructor, in which case a separate
 * thread will be used to drain to the queue when it is close to capacity.
 *
 * @since 1.0.0-beta
 */
public class QueueingScanner extends Scanner {

	private final BlockingQueue<String> queue;

	private BlockingQueue<String> slow = null;
	private ExecutorService executor = null;
	private int threshold = 0;
	private volatile boolean draining = false;

	/**
	 * Creates a {@code QueueingScanner} that sends all results from the
	 * {@link Scanner} straight to the underlying {@link BlockingQueue}.
	 *
	 * @param logger logger
	 * @param queue results from the scanner will be put on this queue
	 */
	public QueueingScanner(final Logger logger, final BlockingQueue<String> queue) {
		super(logger);
		this.queue = queue;
	}

	/**
	 * Creates a {@code QueueingScanner} that buffers all results from the
	 * the scanner in an {@link ArrayBlockingQueue} with the given capacity.
	 *
	 * When the internal queue is nearing capacity, a separate thread is
	 * started which drains it to the underlying queue.
	 *
	 * Use this constructor when the underlying queue suffers from latency
	 * that would slow down an otherwise fast directory scanning operation.
	 *
	 * @param logger logger
	 * @param queue results from the scanner will be put on this queue
	 */
	public QueueingScanner(final Logger logger, final BlockingQueue<String> queue,
		final int buffer) {
		this(logger, new ArrayBlockingQueue<String>(buffer));
		this.slow = queue;
		this.executor = Executors.newSingleThreadExecutor();
		this.threshold = (int) Math.ceil((double) buffer / 1.6);
	}

	@Override
	protected void handle(final Path file) {
		if (null != slow && !draining && queue.size() > threshold) {
			draining = true;
			executor.submit(new DrainingTask());
		}

		try {
			queue.put(file.toString());
		} catch (InterruptedException e) {
			logger.warning("Scanner interrupted while waiting for a free queue slot.");
			Thread.currentThread().interrupt();
		}
	}

	/**
	 * Runnable task that drains the internal queue to an underlying queue
	 * which is where we actually want the file paths to go.
	 */
	protected class DrainingTask implements Runnable {

		@Override
		public void run() {
			logger.info("Scanner draining to slow queue.");
			queue.drainTo(slow);
			draining = false;
		}
	}
}

@@ -199,16 +199,66 @@
 Service;
+%0Aimport java.util.concurrent.atomic.AtomicBoolean;
 %0A%0Aimport
@@ -798,16 +798,73 @@
 ity.%0A *%0A
+ * In both cases, this implementation is thread-safe.%0A *%0A
  * @sinc
@@ -1072,69 +1072,65 @@
 ate 
-int threshold = 0;%0A%09private volatile boolean draining = false
+AtomicBoolean draining = null;%0A%09private int threshold = 0
 ;%0A%0A%09
@@ -2263,16 +2263,55 @@
 utor();%0A
+%09%09this.draining = new AtomicBoolean();%0A
 %09%09this.t
@@ -2443,21 +2443,8 @@
 w &&
- !draining &&
  que
@@ -2464,19 +2464,19 @@
 hreshold
-) %7B
+ &&
 %0A%09%09%09drai
@@ -2483,16 +2483,38 @@
 ning
- =
+.compareAndSet(false,
  true
-;
+)) %7B
 %0A%09%09%09
@@ -3074,16 +3074,19 @@
 ning
- = 
+.set(
 false
+)
 ;%0A%09%09

Make queueing scanner thread-safe, even though it's not use in a multi-threaded context yet