D-Net

package eu.dnetlib.parthenos.workflows.nodes;

import java.net.URI;

import java.net.URISyntaxException;

import java.util.List;

import java.util.Map;

import java.util.NoSuchElementException;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Queues;

import eu.dnetlib.data.collector.ThreadSafeIterator;

import eu.dnetlib.rmi.data.CollectorServiceRuntimeException;

import org.apache.commons.lang3.StringEscapeUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.http.client.utils.URIBuilder;

import org.springframework.http.*;

import org.springframework.web.client.ResourceAccessException;

import org.springframework.web.client.RestClientException;

import org.springframework.web.client.RestTemplate;

/**

 * Created by Alessia Bardi on 31/01/2018.

 *

 * @author Alessia Bardi

 */

public class VirtuosoParthenosIterator extends ThreadSafeIterator {

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

	protected static final int QUEUE_TIMEOUT_SECONDS = 100;

	public final static String TERMINATOR = "ARNOLD";

	public final static String ERROR_TERMINATOR = "SCHWARZ";

	protected final static int SLEEP_MS = 5000;

	protected final static int MAX_RETRIES = 3;

	protected final static int LIMIT = 100;

	private String datasourceName;

	private String datasourceInterface;

	private String virtuosoReaderAPIUrl;

	private boolean started = false;

	private Map<String, Integer> errors = Maps.newHashMap();

	private List<String> listForClass = Lists.newArrayList();

	private BlockingQueue<String> elements = Queues.newArrayBlockingQueue(10);

	private String currentElement = null;

	private ExecutorService executor = Executors.newSingleThreadExecutor();

	private RestTemplate restTemplate;

	private synchronized void verifyStarted(){

		if (!this.started) {

			this.started = true;

			fillQueue();

			getNextElement(MAX_RETRIES);

		}

	}

	protected void fillQueue(){

		log.info("Virtuoso reader at : " + getVirtuosoReaderAPIUrl());

		getExecutor().submit(() -> {

			try {

				int offset = 0;

				boolean again = true;

				do {

					List<String> subjectList = getSubjectList(offset);

					for (String subject : subjectList) {

						String xmlFile = tryGetRDF(subject, MAX_RETRIES);

						if (StringUtils.isBlank(xmlFile)) {

							log.warn("Skipping blank RDF for " + subject);

						} else {

							getElements().offer(xmlFile, QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);

						}

					}

					again = subjectList.size() == LIMIT;

					offset += LIMIT;

				} while(again);

				log.debug("End of subject list, adding terminator to the queue");

				getElements().offer(TERMINATOR, QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);

			} catch (Exception e) {

				log.error(e.getMessage());

				try {

					getElements().offer(ERROR_TERMINATOR, QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);

				} catch (InterruptedException e1) {

					log.error(e1.getMessage());

				}

			}

		});

		getExecutor().shutdown();

	}

	protected String tryGetRDF(final String subjectURL, int attempt) throws URISyntaxException, InterruptedException, VirtuosoParthenosException {

		log.debug("Quering Api, remaining attempts: "+attempt);

		if (attempt <= 0) {

			errors.merge("Failed tryGetRDF", 1, Integer::sum);

			return null;

		}

		ResponseEntity<String> response = null;

		try {

			response = getRDF(subjectURL);

		}catch(ResourceAccessException e){

			//request timed out --> sleep and try again

			log.warn("Request timeout for "+subjectURL+": I'll sleep and then try again");

			Thread.sleep(SLEEP_MS);

			return tryGetRDF(subjectURL, --attempt);

		}

		HttpStatus responseStatus = response.getStatusCode();

		if (responseStatus.is2xxSuccessful()) {

			String rdfFile = response.getBody();

			if(StringUtils.isBlank(rdfFile)){

				log.warn("Got blank RDF for "+subjectURL+" , let's try again...");

				Thread.sleep(SLEEP_MS);

				return tryGetRDF(subjectURL, --attempt);

			}

			else {

				final String xmlFile = completeXML(rdfFile, subjectURL);

				return xmlFile;

			}

		} else {

			if (responseStatus.is5xxServerError()) {

				//sleep for a while and re-try

				log.warn("HTTP ERROR: "+responseStatus.value() + ": " + responseStatus.getReasonPhrase()+": I'll sleep and then try again");

				Thread.sleep(SLEEP_MS);

				return tryGetRDF(subjectURL, --attempt);

			} else {

				log.error("ERROR: Can't get the RDF for " + subjectURL + " " + responseStatus.value() + ": " + responseStatus.getReasonPhrase());

				errors.merge(responseStatus.value() + ": " + responseStatus.getReasonPhrase(), 1, Integer::sum);

			}

		}

		return null;

	}

	protected URI getURIForSubjectList(final int offset) throws URISyntaxException {

		URIBuilder builder = new URIBuilder(getVirtuosoReaderAPIUrl() + "/apiSubjects");

		builder.addParameter("api", getDatasourceInterface());

		builder.addParameter("limit", Integer.toString(LIMIT));

		builder.addParameter("offset", Integer.toString(offset));

		return builder.build();

	}

	protected List<String> getSubjectList(final int offset) throws URISyntaxException, VirtuosoParthenosException {

		URI uri = getURIForSubjectList(offset);

		log.debug("fillQueue -- Calling for subject list: " + uri.toString());

		List<String> subjectList;

		try {

			subjectList = getRestTemplate().getForObject(uri, getListForClass().getClass());

		}catch(RestClientException rce){

			throw new VirtuosoParthenosException(rce);

		}

		return subjectList;

	}

	protected URI getURIForRDFRequest(final String subjectURL) throws URISyntaxException {

		URIBuilder builder = new URIBuilder(getVirtuosoReaderAPIUrl() + "/subject").addParameter("subjectURL", subjectURL);

		return builder.build();

	}

	protected ResponseEntity<String> getRDF(final String subjectURL) throws URISyntaxException {

		HttpHeaders headers = new HttpHeaders();

		headers.setAccept(Lists.newArrayList(MediaType.APPLICATION_XML));

		URI uri = getURIForRDFRequest(subjectURL);

		log.debug("fillQueue -- Calling for subject RDF: " + uri.toString());

		HttpEntity<String> entity = new HttpEntity<>("parameters", headers);

		return restTemplate.exchange(uri, HttpMethod.GET, entity, String.class);

	}

	public String completeXML(final String rdfFile, final String url) {

		String xmlEscapedURL = StringEscapeUtils.escapeXml11(url);

		String rdfFileNoXmlDecl = rdfFile.replaceAll("\\<\\?xml(.+?)\\?\\>", "").trim();

		return "<?xml version=\"1.0\" encoding=\"UTF-8\"?><record xmlns=\"http://www.openarchives.org/OAI/2.0/\"><header xmlns:dri=\"http://www.driver-repository.eu/namespace/dri\"><dri:objIdentifier>"

				+ xmlEscapedURL + "</dri:objIdentifier><dri:datasourceapi>" + datasourceInterface + "</dri:datasourceapi><dri:datasourcename>" + datasourceName

				+ "</dri:datasourcename></header><metadata>" + rdfFileNoXmlDecl + "</metadata></record>";

	}

	@Override

	public boolean doHasNext() {

		try {

			verifyStarted();

		} catch (Exception e) {

			getExecutor().shutdownNow();

			throw new CollectorServiceRuntimeException(e);

		}

		switch(currentElement){

		case TERMINATOR:

			 if(!executor.isTerminated()) executor.shutdownNow();

			return false;

		case ERROR_TERMINATOR:

			 executor.shutdownNow();

			throw new CollectorServiceRuntimeException("Error getting elements from virtuoso");

		default:

			return true;

		}

	}

	@Override

	public String doNext() {

		if(!hasNext()) {

			log.error("Next called but hasNext is false", new NoSuchElementException());

			throw new NoSuchElementException();

		}

		switch(currentElement){

		case TERMINATOR:

		case ERROR_TERMINATOR:

			executor.shutdownNow();

			throw new NoSuchElementException();

		default:

			String res = currentElement;

			getNextElement(MAX_RETRIES);

			return res;

		}

	}

	private void getNextElement(int attempt){

		log.debug("polling from queue, remaining attempts: "+attempt);

		if(attempt <= 0) currentElement = ERROR_TERMINATOR;

		else{

			try {

				currentElement = elements.take();

			} catch (InterruptedException e) {

				//current thread interrupted. Let's end.

				currentElement = ERROR_TERMINATOR;

				executor.shutdownNow();

			}

		}

	}

	@Override

	public void remove() {

		throw new UnsupportedOperationException();

	}

	public VirtuosoParthenosIterator datasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

		return this;

	}

	public VirtuosoParthenosIterator virtuosoReaderAPIUrl(final String virtuosoReaderAPIUrl) {

		this.virtuosoReaderAPIUrl = virtuosoReaderAPIUrl;

		return this;

	}

	public VirtuosoParthenosIterator datasourceName(final String datasourceName) {

		this.datasourceName = datasourceName;

		return this;

	}

	public VirtuosoParthenosIterator errors(final Map<String, Integer> errors) {

		this.errors = errors;

		return this;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public String getVirtuosoReaderAPIUrl() {

		return virtuosoReaderAPIUrl;

	}

	public Map<String, Integer> getErrors() {

		return errors;

	}

	public BlockingQueue<String> getElements() {

		return elements;

	}

	public RestTemplate getRestTemplate() {

		return restTemplate;

	}

	public VirtuosoParthenosIterator restTemplate(final RestTemplate restTemplate) {

		this.restTemplate = restTemplate;

		return this;

	}

	public String getDatasourceName() {

		return datasourceName;

	}

	public boolean isStarted() {

		return started;

	}

	public List<String> getListForClass() {

		return listForClass;

	}

	public String getCurrentElement() {

		return currentElement;

	}

	public ExecutorService getExecutor() {

		return executor;

	}

	public void setDatasourceName(final String datasourceName) {

		this.datasourceName = datasourceName;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public void setVirtuosoReaderAPIUrl(final String virtuosoReaderAPIUrl) {

		this.virtuosoReaderAPIUrl = virtuosoReaderAPIUrl;

	}

	public void setStarted(final boolean started) {

		this.started = started;

	}

	public void setErrors(final Map<String, Integer> errors) {

		this.errors = errors;

	}

	public void setListForClass(final List<String> listForClass) {

		this.listForClass = listForClass;

	}

	public void setElements(final BlockingQueue<String> elements) {

		this.elements = elements;

	}

	public void setCurrentElement(final String currentElement) {

		this.currentElement = currentElement;

	}

	public void setExecutor(final ExecutorService executor) {

		this.executor = executor;

	}

	public void setRestTemplate(final RestTemplate restTemplate) {

		this.restTemplate = restTemplate;

	}

}