D-Net

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

import eu.dnetlib.data.collector.plugins.schemaorg.RepositoryIterable;

import eu.dnetlib.data.collector.plugins.schemaorg.RepositoryQueueIterator;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import java.net.URL;

import java.util.Iterator;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class SitemapIndexRepositoryIterable implements RepositoryIterable {

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

	public static class Options {

		private SitemapIndexIterator.Options sitemapIndexIteratorOptions;

		private SitemapFileIterator.Options sitemapFileIteratorOptions;

		private RepositoryQueueIterator.Options repositoryQueueIteratorOptions;

		private int queueSize;

		public int getQueueSize() {

			return queueSize;

		}

		public void setQueueSize(int queueSize) {

			this.queueSize = queueSize;

		}

		public RepositoryQueueIterator.Options getRepositoryQueueIteratorOptions() {

			return repositoryQueueIteratorOptions;

		}

		public void setRepositoryQueueIteratorOptions(RepositoryQueueIterator.Options repositoryQueueIteratorOptions) {

			this.repositoryQueueIteratorOptions = repositoryQueueIteratorOptions;

		}

		public SitemapIndexIterator.Options getSitemapIndexIteratorOptions() {

			return sitemapIndexIteratorOptions;

		}

		public void setSitemapIndexIteratorOptions(SitemapIndexIterator.Options sitemapIndexIteratorOptions) {

			this.sitemapIndexIteratorOptions = sitemapIndexIteratorOptions;

		}

		public SitemapFileIterator.Options getSitemapFileIteratorOptions() {

			return sitemapFileIteratorOptions;

		}

		public void setSitemapFileIteratorOptions(SitemapFileIterator.Options sitemapFileIteratorOptions) {

			this.sitemapFileIteratorOptions = sitemapFileIteratorOptions;

		}

	}

	private Options options;

	private ArrayBlockingQueue<String> queue;

	public SitemapIndexRepositoryIterable(Options options) {

		this.options = options;

	}

	public void bootstrap() {

		this.queue = new ArrayBlockingQueue<>(this.options.getQueueSize());

		Thread ft = new Thread(new Harvester() );

		ft.start();

//		ExecutorService executor = Executors.newSingleThreadExecutor();

//		executor.execute(new Harvester());

//		executor.shutdown();

	}

	@Override

	public Iterator<String> iterator() {

		return new RepositoryQueueIterator(this.options.getRepositoryQueueIteratorOptions(), this.queue);

	}

	private class Harvester implements Runnable{

		@Override

		public void run() {

			this.execute();

		}

		private void execute(){

			try {

				SitemapIndexIterator sitemapIndexIterator = new SitemapIndexIterator(options.getSitemapIndexIteratorOptions());

				sitemapIndexIterator.bootstrap();

				while (sitemapIndexIterator.hasNext()) {

					String sitemapFile = sitemapIndexIterator.next();

					if(sitemapFile == null) continue;

					SitemapFileIterator.Options sitemapFileIteratorOptions = (SitemapFileIterator.Options)options.getSitemapFileIteratorOptions().clone();

					sitemapFileIteratorOptions.setFileUrl(new URL(sitemapFile));

					SitemapFileIterator sitemapFileIterator = new SitemapFileIterator(sitemapFileIteratorOptions);

					sitemapFileIterator.bootstrap();

					while(sitemapFileIterator.hasNext()){

						String endpoint = sitemapFileIterator.next();

						if(endpoint == null) continue;;

						log.debug("adding endpoint in queue");

						log.debug("queue size: " + queue.size());

						queue.put(endpoint);

						log.debug("endpoint added in queue");

						log.debug("queue size: " + queue.size());

					}

				}

			}catch(Exception ex){

				log.error("problem execution harvesting", ex);

			}

			finally {

				try {

					queue.put(RepositoryIterable.TerminationHint);

				} catch (Exception ex) {

					log.fatal("could not add termination hint. the process will not terminate gracefully", ex);

				}

			}

		}

	}

}