package org.icij.extract.core;

import org.icij.extract.interval.TimeDuration;

import java.util.logging.Logger;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.BlockingQueue;

import java.nio.file.Path;

/**
 * An implementation of {@link Consumer} which polls a given queue for paths to consume.
 * The queue should contain paths serialized to {@link String} objects.
 *
 * When polling is started using {@link start}, the consumer will automatically saturate
 * the thead pool with tasks. The consumer continues to poll in a loop so that the thread
 * pool remains saturated until the queue starts to drain off.
 *
 * @since 1.0.0-beta
 */
public class PollingConsumer extends Consumer {
	public static final TimeDuration DEFAULT_TIMEOUT = new TimeDuration(500L, TimeUnit.MILLISECONDS);

	private final BlockingQueue<String> queue;

	private volatile boolean started = false;
	private TimeDuration pollTimeout = DEFAULT_TIMEOUT;

	public PollingConsumer(Logger logger, BlockingQueue<String> queue, Spewer spewer, Extractor extractor, int threads) {
		super(logger, spewer, extractor, threads);
		this.queue = queue;
	}

	public void setPollTimeout(TimeDuration pollTimeout) {
		this.pollTimeout = pollTimeout;
	}

	public void setPollTimeout(String duration) throws IllegalArgumentException {
		setPollTimeout(new TimeDuration(duration));
	}

	/**
	 * Start consuming.
	 */
	public void start() {
		logger.info("Starting consumer.");

		if (started) {
			throw new IllegalStateException("Already started.");
		}

		started = true;

		String file;

		while (started && null != (file = poll())) {
			consume(file);
		}

		drained();
	}

	/**
	 * Stop consuming.
	 */
	public void stop() {
		started = false;
	}

	protected void drained() {
		logger.info("Queue drained.");
	}

	protected String poll() {
		logger.info(String.format("Polling the queue, waiting up to %s.", pollTimeout));

		String file = null;

		try {
			file = queue.poll(pollTimeout.getDuration(), pollTimeout.getUnit());
		} catch (InterruptedException e) {
			logger.info("Thread interrupted while waiting to poll.");
			Thread.currentThread().interrupt();
		}

		return file;
	}

	protected class ConsumerTask extends Consumer.ConsumerTask {

		public ConsumerTask(final Path file) {
			super(file);
		}

		protected int extract(final Path file) {
			final int status = super.extract(file);

			// If the file could not be written due to a storage endpoint error, put it back onto the queue.
			if (Reporter.NOT_SAVED == status) {
				queue.add(file.toString());
			}

			return status;
		}
	}
}

@@ -777,10 +777,8 @@
 on(5
-00
 L, T
@@ -789,13 +789,8 @@
 nit.
-MILLI
 SECO
@@ -1997,16 +1997,159 @@
 nit());%0A
+%09%09%7D catch (NullPointerException e) %7B%0A%0A%09%09%09// This is a temporary bodge for:%0A%09%09%09// https://github.com/mrniko/redisson/issues/181%0A%09%09%09return file;%0A
 %09%09%7D catc

Fix two bugs caused originating from Redisson: 1) a timeout of less than 1 second is rounded down to 0 which causes the queue to block forever 2) poll with a timeout throws an NPE when the queue is drained