D-Net

package eu.dnetlib.validator.service.impls.providers;

import java.io.IOException;

import java.io.StringReader;

import java.net.URL;

import java.net.URLEncoder;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.Set;

import javax.xml.parsers.DocumentBuilder;

import javax.xml.parsers.DocumentBuilderFactory;

import javax.xml.parsers.ParserConfigurationException;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpression;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import net.sf.ehcache.Cache;

import org.apache.log4j.Logger;

import org.w3c.dom.Document;

import org.w3c.dom.Element;

import org.w3c.dom.Node;

import org.w3c.dom.NodeList;

import org.w3c.dom.ls.DOMImplementationLS;

import org.w3c.dom.ls.LSSerializer;

import org.xml.sax.InputSource;

import org.xml.sax.SAXException;

import se.kb.oai.pmh.OaiPmhServer;

import se.kb.oai.pmh.ResumptionToken;

import se.kb.oai.pmh.SetsList;

import eu.dnetlib.validator.engine.data.DataException;

import eu.dnetlib.validator.engine.data.Provider;

import eu.dnetlib.validator.engine.data.ResultSet;

import eu.dnetlib.validator.engine.execution.ValidationObject;

import eu.dnetlib.validator.service.impls.valobjs.XMLTextValidationObject;

/**

 * A provider that retrieves records from an OAI-PMH repository. Resumption

 * tokens are handled transparently.

 * 

 * @author Nikon Gasparis

 * 

 */

public class CrisProvider extends Provider {

	/**

	 * 

	 */

	private static final long serialVersionUID = -8496954108693674745L;

	private String baseUrl;

	private String metadataPrefix;

	private String set;

	private String records;

	private String from;

	private String until;

	private int timeout;

	private int delay;

	private int retryDelay;

	private int retryEfforts;

	private Cache cache = null;

	private Set<String> entities;

	private Map<String, OAIPMHResultSet> entityResultSetMap = new HashMap<String, OAIPMHResultSet>();

	public CrisProvider() {

		super(4);

	}

	@SuppressWarnings("unchecked")

	@Override

	public synchronized ResultSet<ValidationObject> getValidationObjects(String entity) throws ProviderException {

		if (!entityResultSetMap.containsKey(entity))

		{	

			OAIPMHRecordResultSet resultSet = new OAIPMHRecordResultSet();

			resultSet.setProvider(this);

			resultSet.setValidationSet(entity);

			this.entityResultSetMap.put(entity, resultSet);

		}

		return (ResultSet<ValidationObject>) entityResultSetMap.get(entity);

	}

	public void restartResultSets() {

		Map<String, OAIPMHResultSet> newEntityResultSetMap = new HashMap<String, CrisProvider.OAIPMHResultSet>();

		for (Entry<String, OAIPMHResultSet> ent : entityResultSetMap.entrySet()) {

			OAIPMHReferentialRecordResultSet resultSet = new OAIPMHReferentialRecordResultSet();

			resultSet.setProvider(this);

			resultSet.setValidationSet(ent.getKey());

			newEntityResultSetMap.put(ent.getKey(), resultSet);

		}

		 entityResultSetMap = null;

		 entityResultSetMap = newEntityResultSetMap;

	}

	@Override

	public ResultSet<String> getValidationObjectIds() throws ProviderException {

		return new OAIPMHRecordIdentifierResultSet();

	}

	@Override

	public synchronized ValidationObject getValidationObject(String valObjId) throws ProviderException {

		if (cache != null) {

			net.sf.ehcache.Element element = cache.get(valObjId);

			if (element != null) {

				log.debug("fetching from cache..");

				return (ValidationObject) element.getObjectValue();

			} else {

				log.debug("fetching from server..");

				ValidationObject ret = this.fetchValidationObject(valObjId);

//				net.sf.ehcache.Element ele = new net.sf.ehcache.Element(valObjId, ret);

//				cache.put(ele);

				return ret;

			}

		} else

			return this.fetchValidationObject(valObjId);

	}

	@Override

	public ResultSet<ValidationObject> getValidationObjects()

			throws ProviderException {

		// TODO Auto-generated method stub

		return null;

	}

	public synchronized ValidationObject fetchValidationObject(String valObjId) throws ProviderException {

		log.debug("fetching object with id: " + valObjId);

		try {

			OAIPMHResultSet oai = new OAIPMHResultSet();

			return new XMLTextValidationObject(oai.getRecord(valObjId));

		} catch (DataException e) {

			log.error("error fetching object with id: " + valObjId, e);

			throw new ProviderException();

		}

	}

	private class OAIPMHResultSet {

		Logger log = Logger.getLogger(OAIPMHResultSet.class);

		protected DocumentBuilderFactory factory = DocumentBuilderFactory

				.newInstance();

		protected DocumentBuilder builder;

		protected XPathFactory xfactory = XPathFactory.newInstance();

		protected NodeList recordIds = null;

		protected NodeList records = null;

		protected String validationSet;

		protected int index = -1;

		protected String resumptionToken = null;

		protected String error = null;

		private URLStreamer streamer = new URLStreamer();

		private CrisProvider provider;

		public void setValidationSet(String validationSet) {

			this.validationSet = validationSet;

		}

		public void setProvider(CrisProvider provider) {

			this.provider = provider;

		}

		public OAIPMHResultSet() {

			try {

				builder = factory.newDocumentBuilder();

			} catch (ParserConfigurationException e) {

				log.error("", e);

			}

		}

		protected NodeList getIds() throws DataException {

			NodeList recordIds = null;

			String surl = provider.getBaseUrl() + "?verb=ListIdentifiers";

			if (resumptionToken == null || resumptionToken.trim().length() == 0) {

				surl += "&metadataPrefix=" + provider.getMetadataPrefix();

				if (provider.getFrom() != null)

					surl += "&from=" + provider.getFrom();

				if (provider.getUntil() != null)

					surl += "&until=" + provider.getUntil();

				if (!validationSet.equals("none"))

					surl += "&set=" + validationSet;

			} else {

				surl += "&resumptionToken=" + resumptionToken;

			}

			log.debug("Issuing request " + surl);

			String response = null;

			try {

				response = streamer.getResponse(new URL(surl),

						provider.getTimeout(),

						provider.getDelay(),

						provider.getRetryDelay(),

						provider.getRetryEfforts());

			} catch (Exception e) {

				log.error("", e);

				throw new DataException();

			}

			try {

				InputSource is = new InputSource(new StringReader(response));

				Document doc = builder.parse(is);

				XPath xpath = xfactory.newXPath();

				XPathExpression expr = xpath

						.compile("OAI-PMH/ListIdentifiers/header/identifier/text()");

				recordIds = (NodeList) expr.evaluate(doc, XPathConstants.NODESET);

				XPath xpath2 = xfactory.newXPath();

				XPathExpression expr2 = xpath2

						.compile("OAI-PMH/ListIdentifiers/resumptionToken/text()");

				NodeList rtl = (NodeList) expr2.evaluate(doc,

						XPathConstants.NODESET);

				if (rtl == null || rtl.getLength() == 0) {

					log.debug("There seems to be no resumption token present");

					resumptionToken = null;

				} else {

					resumptionToken = rtl.item(0).getNodeValue();

					resumptionToken = URLEncoder.encode(resumptionToken, "UTF-8");

					log.debug("Found resumption token: " + resumptionToken);

				}

			} catch (Exception e) {

				log.error("", e);

				throw new DataException();

			}

			return recordIds;

		}

		protected NodeList getRecords() throws DataException {

			NodeList records = null;

			String surl = provider.getBaseUrl() + "?verb=ListRecords";

			if (resumptionToken == null || resumptionToken.trim().length() == 0) {

				surl += "&metadataPrefix=" + provider.getMetadataPrefix();

				if (provider.getFrom() != null)

					surl += "&from=" + provider.getFrom();

				if (provider.getUntil() != null)

					surl += "&until=" + provider.getUntil();

				if (!validationSet.equals("none"))

					surl += "&set=" + validationSet;

			} else {

				surl += "&resumptionToken=" + resumptionToken;

			}

			String setError = null;

			if (!validationSet.equals("none")) {

				OaiPmhServer harvester = new OaiPmhServer(provider.getBaseUrl());

				try {

					SetsList setList = harvester.listSets();

					ResumptionToken token = setList.getResumptionToken();

					List<se.kb.oai.pmh.Set> sets = new ArrayList<se.kb.oai.pmh.Set>();

					sets.addAll(setList.asList());

					while (token != null) {

						setList = harvester.listSets(token);

						token = setList.getResumptionToken();

						sets.addAll(setList.asList());

					}

					List<String> ret = new ArrayList<String>();

					for (se.kb.oai.pmh.Set set : sets) {

						ret.add(set.getSpec().trim());

					}

					if (!ret.contains(validationSet)) {

						error = "Set: <b>'"

								+ validationSet

								+ "'</b> is not exposed by the repository. \n Please make sure that 'ListSets' verb is configured correctly on your server as well as that the exposed sets list includes this set.";

						setError = error;

					}

				} catch (Exception e) {

					log.error(

							"error getting sets from url: "

									+ provider.getBaseUrl(), e);

				}

			}

			log.debug("Issuing request " + surl);

			String response = null;

			try {

				response = streamer.getResponse(new URL(surl),

						provider.getTimeout(),

						provider.getDelay(),

						provider.getRetryDelay(),

						provider.getRetryEfforts());

			} catch (Exception e) {

				log.error("", e);

				throw new DataException();

			}

			try {

				InputSource is = new InputSource(new StringReader(response));

				Document doc = builder.parse(is);

				XPath xpath = xfactory.newXPath();

				// xpath.setNamespaceContext(new NamespaceResolver(doc));

				XPathExpression expr = xpath.compile("OAI-PMH/ListRecords/record");

				records = (NodeList) expr.evaluate(doc, XPathConstants.NODESET);

				XPath xpath2 = xfactory.newXPath();

				XPathExpression expr2 = xpath2

						.compile("OAI-PMH/ListRecords/resumptionToken/text()");

				NodeList rtl = (NodeList) expr2.evaluate(doc,

						XPathConstants.NODESET);

				log.debug("Check number of records: " + records.getLength());

				if (records.getLength() == 0) {

					log.debug("There are no records: " + records.getLength());

					XPath xpath3 = xfactory.newXPath();

					XPathExpression expr3 = xpath3.compile("OAI-PMH/error/text()");

					error = "The response on request: <a href=\"" + surl + "\">"

							+ surl + "</a> was the following: ";

					error += "<br><b>"

							+ (String) expr3.evaluate(doc, XPathConstants.STRING)

							+ "</b>";

					if (setError != null) {

						error += "<br>" + setError;

					}

					log.debug("Error: " + error);

				}

				if (rtl == null || rtl.getLength() == 0) {

					log.debug("There seems to be no resumption token present");

					resumptionToken = null;

				} else {

					resumptionToken = rtl.item(0).getNodeValue();

					resumptionToken = URLEncoder.encode(resumptionToken, "UTF-8");

					log.debug("Found resumption token: " + resumptionToken);

				}

			} catch (Exception e) {

				log.error("", e);

				throw new DataException();

			}

			return records;

		}

		protected Document getRecord(String id) throws DataException {

			String surl = provider.getBaseUrl() + "?verb=GetRecord";

			surl += "&metadataPrefix=" + provider.getMetadataPrefix();

			surl += "&identifier=" + id;

			log.debug("Issuing request: " + surl);

			String response = null;

			try {

				response = streamer.getResponse(new URL(surl),

						provider.getTimeout(),

						provider.getDelay(),

						provider.getRetryDelay(),

						provider.getRetryEfforts());

			} catch (Exception e) {

				log.error("", e);

				throw new DataException();

			}

			InputSource is = new InputSource(new StringReader(response));

			Document doc;

			try {

				doc = builder.parse(is);

			} catch (SAXException e) {

				log.error("", e);

				throw new DataException();

			} catch (IOException e) {

				log.error("", e);

				throw new DataException();

			}

			return doc;

		}

	}

	private class OAIPMHRecordResultSet extends OAIPMHResultSet implements ResultSet<ValidationObject> {

		@Override

		public String getError() {

			if (error != null)

				log.debug("An error occured "+ this.error);

			else

				log.debug("No errors on request");

			return this.error;

		}

		@Override

		public boolean next() throws DataException {

			index++;

			log.debug("Moving cursor to result "+index);

			if (records == null || index >= records.getLength()) {

				// if we have previously received some results and there no more to take

				if (records != null && (resumptionToken == null || resumptionToken.trim().length() == 0))

					return false;

				index = -1;

				records = getRecords();

				return next();

			}

			return true;

		}

		@Override

		public ValidationObject get() throws DataException {

			XMLTextValidationObject ret = null;

			Document newXmlDocument;

			try {

				newXmlDocument = DocumentBuilderFactory.newInstance().newDocumentBuilder().newDocument();

	        Element root = newXmlDocument.createElement("root");

	        newXmlDocument.appendChild(root);

	            Node node = records.item(index);

	            Node copyNode = newXmlDocument.importNode(node, true);

	            root.appendChild(copyNode);

//	            printXmlDocument(newXmlDocument);

	            ret = new XMLTextValidationObject(newXmlDocument);

	            XPathFactory factory = XPathFactory.newInstance();

	            XPath xPath = factory.newXPath();

	            ret.setId(xPath.evaluate("header/identifier", records.item(index)));

	            ret.setStatus(xPath.evaluate("header/@status", records.item(index)));

			} catch (ParserConfigurationException e) {

				log.error("error getting object"+ e);

			} catch (XPathExpressionException e) {

				log.error("error getting object"+ e);

			}			

			if (cache != null) {

				log.debug("adding to cache..");

				net.sf.ehcache.Element ele = new net.sf.ehcache.Element(ret.getId(), ret);

				cache.put(ele);

			}

			return ret;

		}

	}

	public static void printXmlDocument(Document document) {

	    DOMImplementationLS domImplementationLS = 

	        (DOMImplementationLS) document.getImplementation();

	    LSSerializer lsSerializer = 

	        domImplementationLS.createLSSerializer();

	    String string = lsSerializer.writeToString(document);

	    System.out.println(string);

	}

	private class OAIPMHRecordIdentifierResultSet extends OAIPMHResultSet implements ResultSet<String> {

		public OAIPMHRecordIdentifierResultSet() {

		}

		@Override

		public boolean next() throws DataException {

			index++;

			log.debug("Moving cursor to result "+index);

			if (recordIds == null || index >= recordIds.getLength()) {

				// if we have previously received some results and there no more to take

				if (recordIds != null && (resumptionToken == null || resumptionToken.trim().length() == 0))

					return false;

				index = -1;

				recordIds = getIds();

				return next();

			}

			return true;

		}

		@Override

		public String get() throws DataException {

			String id = recordIds.item(index).getNodeValue();

			log.debug("Returing object with id "+id);

			return id;

		}

		@Override

		public String getError() {

			// TODO Auto-generated method stub

			return null;

		}

	}

	private class OAIPMHReferentialRecordResultSet extends OAIPMHResultSet implements ResultSet<ValidationObject> {

		public OAIPMHReferentialRecordResultSet() {

		}

		@Override

		public boolean next() throws DataException {

			index++;

			log.debug("Moving cursor to result "+index);

			if (recordIds == null || index >= recordIds.getLength()) {

				// if we have previously received some results and there no more to take

				if (recordIds != null && (resumptionToken == null || resumptionToken.trim().length() == 0))

					return false;

				index = -1;

				recordIds = getIds();

				return next();

			}

			return true;

		}

		@Override

		public ValidationObject get() throws DataException {

			String id = recordIds.item(index).getNodeValue();

			log.debug("Returing object with id "+id);

			net.sf.ehcache.Element element = cache.get(id);

			if (element != null) {

				return (ValidationObject) element.getObjectValue();

			} else {

				return new XMLTextValidationObject(this.getRecord(id));

			}

//			return (ValidationObject) selfPopulatingCache.get(id).getObjectValue();

		}

		@Override

		public String getError() {

			// TODO Auto-generated method stub

			return null;

		}

	}

	public Set<String> getEntities() {

		return entities;

	}

	public Cache getCache() {

		return cache;

	}

	public void setCache(Cache cache) {

		this.cache = cache;

	}

	public void setEntities(Set<String> entities) {

		this.entities = entities;

	}

	public String getBaseUrl() {

		return baseUrl;

	}

	public void setBaseUrl(String baseUrl) {

		this.baseUrl = baseUrl;

	}

	public String getMetadataPrefix() {

		return metadataPrefix;

	}

	public void setMetadataPrefix(String metadataPrefix) {

		this.metadataPrefix = metadataPrefix;

	}

	public String getRecords() {

		return records;

	}

	public void setRecords(String records) {

		this.records = records;

	}

	public String getFrom() {

		return from;

	}

	public void setFrom(String from) {

		this.from = from;

	}

	public String getUntil() {

		return until;

	}

	public void setUntil(String until) {

		this.until = until;

	}

	public String getSet() {

		return set;

	}

	public void setSet(String set) {

		this.set = set;

	}

	public Integer getTimeout() {

		return timeout;

	}

	public void setTimeout(Integer timeout) {

		this.timeout = timeout;

	}

	public Integer getDelay() {

		return delay;

	}

	public void setDelay(Integer delay) {

		this.delay = delay;

	}

	public Integer getRetryDelay() {

		return retryDelay;

	}

	public void setRetryDelay(Integer retryDelay) {

		this.retryDelay = retryDelay;

	}

	public Integer getRetryEfforts() {

		return retryEfforts;

	}

	public void setRetryEfforts(Integer retryEfforts) {

		this.retryEfforts = retryEfforts;

	}

}