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;

	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() throws Exception {

		if (!this.started) {

			this.started = true;

			fillQueue();

			getNextElement(MAX_RETRIES);

		}

	}

	protected void fillQueue(){

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

		executor.submit(() -> {

			String threadName = Thread.currentThread().getName();

			System.out.println("Hello " + threadName);

			URIBuilder builder;

			try {

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

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

				URI uri = builder.build();

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

				List<String> graphList = restTemplate.getForObject(uri, listForClass.getClass());

				for (String graph : graphList) {

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

					builder.addParameter("graph", graph);

					uri = builder.build();

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

					List<String> subjectList =Lists.newArrayList();

					try {

						subjectList = restTemplate.getForObject(builder.build(), listForClass.getClass());

					}catch(RestClientException rce){

						log.warn("Skipping subjects from "+graph+" and sleeping for a while...");

						log.warn(rce.getMessage());

						Thread.sleep(SLEEP_MS);

					}

					for (String subject : subjectList) {

						//Note: we skip no http resources based on FORTH feedback: URLs are created only for "relevant" stuff

						//TODO: is it better to skip them here or in the query run by VirtuosoReadApi?

						if (subject.startsWith("http")) {

							String xmlFile = tryGetRDF(subject, MAX_RETRIES);

							if (StringUtils.isBlank(xmlFile)) {

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

							} else {

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

							}

						} else {

							log.debug("Skipping " + subject + ": no http resource");

						}

					}

				}

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

			} catch (Exception e) {

				log.error(e.getMessage());

				try {

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

				} catch (InterruptedException e1) {

					log.error(e1.getMessage());

				}

			}

		});

		executor.shutdown();

	}

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

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

		if (attempt <= 0) throw new VirtuosoParthenosException("Too many retries...stopping. please check virtuoso server status");

		ResponseEntity<String> response = null;

		try {

			response = getRDF(subjectURL);

		}catch(ResourceAccessException e){

			//request timed out --> let's return null and skip this resource

			log.warn("Request timeout");

			errors.merge(e.getMessage(), 1, Integer::sum);

			return null;

		}

		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);

				log.debug(xmlFile);

				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 {

				throw new VirtuosoParthenosException(

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

			}

		}

	}

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

		HttpHeaders headers = new HttpHeaders();

		headers.setContentType(MediaType.APPLICATION_XML);

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

		URI uri = builder.build();

		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);

		return "<?xml version=\"1.0\" encoding=\"UTF-8\"?><record><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>" + rdfFile + "</metadata></record>";

	}

	@Override

	public boolean doHasNext() {

		try {

			verifyStarted();

		} catch (Exception e) {

			executor.shutdownNow();

			throw new CollectorServiceRuntimeException(e);

		}

		switch(currentElement){

		case TERMINATOR:

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

			return false;

		case ERROR_TERMINATOR:

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

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

		default:

			return true;

		}

	}

	@Override

	public String doNext() {

		try {

			verifyStarted();

		} catch (Exception e) {

			executor.shutdownNow();

			throw new CollectorServiceRuntimeException(e);

		}

		switch(currentElement){

		case TERMINATOR:

		case ERROR_TERMINATOR:

			if(!executor.isShutdown()) 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.poll(QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);

				if(currentElement == null){

					if (executor.isTerminated()) {

						//no additional element and we got the timeout...

						String errMessage = "Unexpected timeout reading element queue";

						log.warn(errMessage);

						errors.merge(errMessage, 1, Integer::sum);

						currentElement = ERROR_TERMINATOR;

					} else {

						//executor is not terminated: let's try again

						log.warn("Queue timed out, but I'll try again (#"+attempt+")");

						getNextElement(--attempt);

					}

				}

			} 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;

	}

}