D-Net

package eu.dnetlib.data.collector.plugins.projects.gtr2;

import java.util.Iterator;

import java.util.NoSuchElementException;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

import com.ximpleware.AutoPilot;

import com.ximpleware.VTDGen;

import com.ximpleware.VTDNav;

import eu.dnetlib.data.collector.rmi.CollectorServiceException;

import eu.dnetlib.data.collector.rmi.CollectorServiceRuntimeException;

import eu.dnetlib.enabling.resultset.SizedIterable;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.joda.time.DateTime;

import org.joda.time.format.DateTimeFormat;

import org.joda.time.format.DateTimeFormatter;

/**

 * Created by alessia on 28/11/16.

 */

public class Gtr2ProjectsIterable implements SizedIterable<String> {

	public static final String TERMINATOR = "ARNOLD";

	public static final int WAIT_END_SECONDS = 120;

	public static final int PAGE_SZIE = 20;

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

	private String queryURL;

	private int total = -1;

	private int startFromPage = 1;

	private int endAtPage;

	private VTDGen vg;

	private VTDNav vn;

	private AutoPilot ap;

	private String namespaces;

	private boolean incremental = false;

	private DateTime fromDate;

	private DateTimeFormatter simpleDateTimeFormatter = DateTimeFormat.forPattern("yyyy-MM-dd");

	private ArrayBlockingQueue<String> projects = new ArrayBlockingQueue<String>(20);

	//private boolean finished = false;

	private final ExecutorService es = Executors.newFixedThreadPool(PAGE_SZIE);

	private String nextElement = null;

	public Gtr2ProjectsIterable(final String baseUrl, final String fromDate) throws CollectorServiceException {

		prepare(baseUrl, fromDate);

		fillInfo(true);

	}

	public Gtr2ProjectsIterable(final String baseUrl, final String fromDate, final int startFromPage, final int endAtPage) throws CollectorServiceException {

		prepare(baseUrl, fromDate);

		this.setStartFromPage(startFromPage);

		this.setEndAtPage(endAtPage);

		fillInfo(false);

	}

	private void prepare(final String baseUrl, final String fromDate) {

		queryURL = baseUrl + "/projects";

		vg = new VTDGen();

		this.incremental = StringUtils.isNotBlank(fromDate);

		if (incremental) {

			// I expect fromDate in the format 'yyyy-MM-dd'. See class eu.dnetlib.msro.workflows.nodes.collect.FindDateRangeForIncrementalHarvestingJobNode

			this.fromDate = DateTime.parse(fromDate, simpleDateTimeFormatter);

			log.debug("fromDate string: " + fromDate + " -- parsed: " + this.fromDate.toString());

		}

	}

	@Override

	public int getNumberOfElements() {

		return total;

	}

	private void fillInfo(final boolean all) throws CollectorServiceException {

		try {

			// log.debug("Getting hit count from: " + queryURL);

			vg.parseHttpUrl(queryURL, false);

			initParser();

			String hitCount = vn.toNormalizedString(vn.getAttrVal("totalSize"));

			String totalPages = vn.toNormalizedString(vn.getAttrVal("totalPages"));

			namespaces = "xmlns:ns1=\"" + vn.toNormalizedString(vn.getAttrVal("ns1")) + "\" ";

			namespaces += "xmlns:ns2=\"" + vn.toNormalizedString(vn.getAttrVal("ns2")) + "\" ";

			namespaces += "xmlns:ns3=\"" + vn.toNormalizedString(vn.getAttrVal("ns3")) + "\" ";

			namespaces += "xmlns:ns4=\"" + vn.toNormalizedString(vn.getAttrVal("ns4")) + "\" ";

			namespaces += "xmlns:ns5=\"" + vn.toNormalizedString(vn.getAttrVal("ns5")) + "\" ";

			namespaces += "xmlns:ns6=\"" + vn.toNormalizedString(vn.getAttrVal("ns6")) + "\" ";

			if (all) {

				setEndAtPage(Integer.parseInt(totalPages));

				total = Integer.parseInt(hitCount);

			}

			Thread ft = new Thread(new FillProjectList());

			ft.start();

			log.debug("Expected number of pages: " + (endAtPage - startFromPage + 1));

		} catch (NumberFormatException e) {

			log.error("Cannot set the total count or the number of pages");

			throw new CollectorServiceException(e);

		} catch (Throwable e) {

			throw new CollectorServiceException(e);

		}

	}

	@Override

	public Iterator<String> iterator() {

		return new Iterator<String>() {

			// The following is for debug only

			private int nextCounter = 0;

			@Override

			public boolean hasNext() {

				try {

					log.debug("hasNext?");

					if (nextElement == null) {

						nextElement = projects.poll(WAIT_END_SECONDS, TimeUnit.SECONDS);

						log.debug("Exit poll :-)");

					}

					return nextElement != null && !nextElement.equals(TERMINATOR);

				} catch (InterruptedException e) {

					throw new CollectorServiceRuntimeException(e);

				}

			}

			@Override

			public String next() {

				nextCounter++;

				log.debug(String.format("Calling next %s times.", nextCounter));

				if (nextElement == null) throw new NoSuchElementException();

				else {

					String res = nextElement;

					nextElement = null;

					return res;

				}

			}

			@Override

			public void remove() {

				throw new UnsupportedOperationException();

			}

		};

	}

	private void initParser() {

		vn = vg.getNav();

		ap = new AutoPilot(vn);

	}

	public String getQueryURL() {

		return queryURL;

	}

	public void setQueryURL(final String queryURL) {

		this.queryURL = queryURL;

	}

	public int getTotal() {

		return total;

	}

	public void setTotal(final int total) {

		this.total = total;

	}

	public int getEndAtPage() {

		return endAtPage;

	}

	public void setEndAtPage(final int endAtPage) {

		this.endAtPage = endAtPage;

		log.debug("Overriding endAtPage to " + endAtPage);

	}

	public VTDGen getVg() {

		return vg;

	}

	public void setVg(final VTDGen vg) {

		this.vg = vg;

	}

	public VTDNav getVn() {

		return vn;

	}

	public void setVn(final VTDNav vn) {

		this.vn = vn;

	}

	public AutoPilot getAp() {

		return ap;

	}

	public void setAp(final AutoPilot ap) {

		this.ap = ap;

	}

	public String getNamespaces() {

		return namespaces;

	}

	public void setNamespaces(final String namespaces) {

		this.namespaces = namespaces;

	}

	public int getStartFromPage() {

		return startFromPage;

	}

	public void setStartFromPage(final int startFromPage) {

		this.startFromPage = startFromPage;

		log.debug("Overriding startFromPage to " + startFromPage);

	}

	private class FillProjectList implements Runnable {

		private boolean morePages = true;

		private int pageNumber = startFromPage;

		@Override

		public void run() {

			String resultPageUrl = "";

			try {

				do {

					resultPageUrl = getNextPageUrl();

					log.debug("Page: " + resultPageUrl);

					// clear VGen before processing the next file

					vg.clear();

					vg.parseHttpUrl(resultPageUrl, false);

					initParser();

					ap.selectXPath("//project");

					int res;

					while ((res = ap.evalXPath()) != -1) {

						final String projectHref = vn.toNormalizedString(vn.getAttrVal("href"));

						Thread t = new Thread(new ParseProject(projectHref));

						t.setName("Thread for " + res);

						es.execute(t);

					}

					ap.resetXPath();

				} while (morePages);

				es.shutdown();

				es.awaitTermination(WAIT_END_SECONDS, TimeUnit.SECONDS);

				projects.put(TERMINATOR);

			} catch (Throwable e) {

				log.error("Exception processing " + resultPageUrl + "\n" + e.getMessage());

			}

		}

		private String getNextPageUrl() {

			String url = queryURL + "?p=" + pageNumber;

			if (pageNumber == endAtPage) {

				morePages = false;

			}

			pageNumber++;

			return url;

		}

	}

	private class ParseProject implements Runnable {

		VTDNav vn1;

		VTDGen vg1;

		private String projectRef;

		public ParseProject(final String projectHref) {

			projectRef = projectHref;

			vg1 = new VTDGen();

			vg1.parseHttpUrl(projectRef, false);

			vn1 = vg1.getNav();

		}

		private int projectsUpdate(String attr) throws CollectorServiceException {

			try {

				int index = vn1.getAttrVal(attr);

				if (index != -1) {

					String d = vn1.toNormalizedString(index);

					DateTime recordDate = DateTime.parse(d.substring(0, d.indexOf("T")), simpleDateTimeFormatter);

					// updated or created after the last time it was collected

					if (recordDate.isAfter(fromDate)) {

						log.debug("New project to collect");

						return index;

					}

					return -1;

				}

				return index;

			} catch (Throwable e) {

				throw new CollectorServiceException(e);

			}

		}

		private String collectProject() throws CollectorServiceException {

			try {

				int p = vn1.getAttrVal("href");

				final String projectHref = vn1.toNormalizedString(p);

				log.debug("collecting project at " + projectHref);

				Gtr2Helper gtr2Helper = new Gtr2Helper();

				String projectPackage = gtr2Helper.processProject(vn1, namespaces);

				return projectPackage;

			} catch (Throwable e) {

				throw new CollectorServiceException(e);

			}

		}

		private boolean add(String attr) throws CollectorServiceException {

			return projectsUpdate(attr) != -1;

		}

		@Override

		public void run() {

			log.debug("Getting project info from " + projectRef);

			try {

				if (!incremental || (incremental && (add("created") || add("updated")))) {

					projects.put(collectProject());

					log.debug("Project enqueued " + projectRef);

				}

			} catch (Throwable e) {

				log.error("Error on ParseProject " + e.getMessage());

				throw new CollectorServiceRuntimeException(e);

			}

		}

	}

}