D-Net

package eu.dnetlib.data.collector.plugins.schemaorg;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import java.util.Iterator;

import java.util.NoSuchElementException;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.TimeUnit;

public class RepositoryQueueIterator implements Iterator<String> {

	private static final Log log = LogFactory.getLog(RepositoryQueueIterator.class);

	public static class Options {

		private Boolean blockPolling;

		private long pollTimeout;

		private TimeUnit pollTimeoutUnit;

		public Boolean getBlockPolling() {

			return blockPolling;

		}

		public void setBlockPolling(Boolean blockPolling) {

			this.blockPolling = blockPolling;

		}

		public long getPollTimeout() {

			return pollTimeout;

		}

		public void setPollTimeout(long pollTimeout) {

			this.pollTimeout = pollTimeout;

		}

		public TimeUnit getPollTimeoutUnit() {

			return pollTimeoutUnit;

		}

		public void setPollTimeoutUnit(TimeUnit pollTimeoutUnit) {

			this.pollTimeoutUnit = pollTimeoutUnit;

		}

	}

	private ArrayBlockingQueue<String> queue;

	private Options options;

	private boolean hasTerminated;

	public RepositoryQueueIterator(Options options, ArrayBlockingQueue<String> queue) {

		this.options = options;

		this.queue = queue;

		this.hasTerminated = false;

	}

	@Override

	public boolean hasNext() {

		if(this.hasTerminated) return false;

		return true;

	}

	@Override

	public String next() {

		String next = this.poll();

		if (next != null && next.equalsIgnoreCase(RepositoryIterable.TerminationHint)) {

			this.hasTerminated = true;

			next = null;

		}

		return next;

	}

	private String poll(){

		if(this.options.getBlockPolling()) {

			try {

				return this.queue.poll(this.options.getPollTimeout(), this.options.getPollTimeoutUnit());

			} catch (InterruptedException ex) {

				log.warn(String.format("could not poll elements from queue for more than %s %s. throwing", this.options.getPollTimeout(), this.options.getPollTimeoutUnit()));

				throw new NoSuchElementException(ex.getMessage());

			}

		}

		else {

			return this.queue.poll();

		}

	}

}