D-Net

/**

 * 

 */

package eu.dnetlib.data.collector.plugins.rest;

import java.io.InputStream;

import java.io.StringWriter;

import java.net.URL;

import java.util.Iterator;

import java.util.Queue;

import java.util.concurrent.PriorityBlockingQueue;

import javax.xml.transform.OutputKeys;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerConfigurationException;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.dom.DOMSource;

import javax.xml.transform.stream.StreamResult;

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 org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.w3c.dom.Node;

import org.w3c.dom.NodeList;

import org.xml.sax.InputSource;

import eu.dnetlib.data.collector.plugins.oai.OaiIterator;

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

/**

 * @author Jochen Schirrwagen, Aenne Loehden, Andreas Czerniak

 * @date 2018-08-06

 *

 */

public class RestIterator implements Iterator<String> {

    // TODO: clean up the comments of replaced source code

	private static final Log log = LogFactory.getLog(OaiIterator.class); // NOPMD by marko on 11/24/08 5:02 PM

	private static final String wrapName = "recordWrap";

	private String baseUrl;

	private String resumptionType;

	private String resumptionParam;

	private String resultFormatValue;

	private String queryParams;

	private int resultSizeValue;

	private int resumptionInt = 0;			// integer resumption token (first record to harvest)

	private int resultTotal = -1;

	private String resumptionStr = Integer.toString(resumptionInt);  // string resumption token (first record to harvest or token scanned from results)

	private InputStream resultStream;

	private Transformer transformer;

	private XPath xpath;

	private String query;

	private XPathExpression xprResultTotalPath;

	private XPathExpression xprResumptionPath;

	private XPathExpression xprEntity;

	private String queryFormat;

	private String querySize;

	private Queue<String> recordQueue = new PriorityBlockingQueue<String>();

	public RestIterator(

			final String baseUrl,

			final String resumptionType,

			final String resumptionParam,

			final String resumptionXpath,

			final String resultTotalXpath,

			final String resultFormatParam,

			final String resultFormatValue,

			final String resultSizeParam,

                        final String resultSizeValueStr,

			final String queryParams,

			final String entityXpath

			) {

		this.baseUrl = baseUrl;

		this.resumptionType = resumptionType;

		this.resumptionParam = resumptionParam;

		this.resultFormatValue = resultFormatValue;

		this.queryParams = queryParams;

                this.resultSizeValue = Integer.valueOf(resultSizeValueStr);

        queryFormat = StringUtils.isNotBlank(resultFormatParam) ? "&" + resultFormatParam + "=" + resultFormatValue : "";

        querySize = StringUtils.isNotBlank(resultSizeParam) ? "&" + resultSizeParam + "=" + resultSizeValueStr : "";

		try {

			initXmlTransformation(resultTotalXpath, resumptionXpath, entityXpath);

		} catch(Exception e) {

			throw new IllegalStateException("xml transformation init failed: " + e.getMessage());

		}

        initQueue();

	}

	private void initXmlTransformation(String resultTotalXpath, String resumptionXpath, String entityXpath) throws TransformerConfigurationException, XPathExpressionException{

		transformer = TransformerFactory.newInstance().newTransformer();

                transformer.setOutputProperty(OutputKeys.INDENT,"yes"); 

                transformer.setOutputProperty("{http://xml.apache.org/xslt}indent-amount","3");

		xpath              = XPathFactory.newInstance().newXPath();

		xprResultTotalPath = xpath.compile(resultTotalXpath);

		xprResumptionPath  = xpath.compile(StringUtils.isBlank(resumptionXpath) ? "/" : resumptionXpath);

		xprEntity          = xpath.compile(entityXpath);

	}

	private void initQueue() {

		query = baseUrl + "?" + queryParams + querySize + "&" + resumptionParam + "=" + resumptionStr + queryFormat;

	}

	private void disconnect() {

		// TODO close inputstream

	}

	/* (non-Javadoc)

	 * @see java.util.Iterator#hasNext()

	 */

	@Override

	public boolean hasNext() {

		if (recordQueue.isEmpty() && query.isEmpty()) {

			disconnect();

			return false;

		} else {

			return true;

		}

	}

	/* (non-Javadoc)

	 * @see java.util.Iterator#next()

	 */

	@Override

	public String next() {

		synchronized (recordQueue) {

			while (recordQueue.isEmpty() && !query.isEmpty() ) {

				try {

					query = downloadPage(query);

				} catch(CollectorServiceException e) {

					throw new RuntimeException(e);

				}

			}

			return recordQueue.poll();

		}

	}

	/*

	 * download page and return nextQuery

	 */

	private String downloadPage(String query) throws CollectorServiceException{

		String resultJson;

		String resultXml = "";

		String nextQuery = "";

                String emptyXml = "<"+wrapName+"></"+wrapName+">";

                Node resultNode = null;

                NodeList nodeList;

		try {

                        resultStream = new URL(query).openStream();

			if("json".equals(resultFormatValue.toLowerCase())){				

				resultJson = IOUtils.toString(resultStream,"UTF-8");

				//TODO move regex definitions as constant fields

				// pre-clean json - rid spaces of element names (misinterpreted as elements with attributes in xml)

				resultJson = syntaxConvertJsonKeyNamens(resultJson);

//                                while(resultJson.matches(".*\"([^\"]*)\\s+([^\"]*)\":.*")){

//					resultJson = resultJson.replaceAll("\"([^\"]*)\\s+([^\"]*)\":", "\"$1_$2\":");

//				}

				org.json.JSONObject jsonObject = new org.json.JSONObject(resultJson);

				resultXml = org.json.XML.toString(jsonObject,wrapName); // wrap xml in single root element

				log.trace(resultXml);

				resultStream = IOUtils.toInputStream(resultXml,"UTF-8");

			}

                        if (!(emptyXml.toLowerCase()).equals(resultXml.toLowerCase())) {

                            resultNode = (Node) xpath.evaluate("/", new InputSource(resultStream), XPathConstants.NODE);

                            nodeList   = (NodeList) xprEntity.evaluate(resultNode, XPathConstants.NODESET);

                            log.debug("nodeList length: " + nodeList.getLength());

                            for (int i = 0; i < nodeList.getLength(); i++) {

                                StringWriter sw = new StringWriter();

                                transformer.transform(new DOMSource(nodeList.item(i)), new StreamResult(sw));

                                recordQueue.add(sw.toString());

                            }

                        }	

			resumptionInt += resultSizeValue;

/*	replaced by switch statement as follow		

                        if("scan".equals(resumptionType.toLowerCase())) { resumptionStr = xprResumptionPath.evaluate(resultNode);}

			if("count".equals(resumptionType.toLowerCase())){ resumptionStr = Integer.toString(resumptionInt); }

*/

                        switch(resumptionType.toLowerCase()) {

                            case "scan":    // read of resumptionToken , evaluate next results, e.g. OAI

                                            resumptionStr = xprResumptionPath.evaluate(resultNode);

                                            break;

                            case "count":   // begin at one step for all records

                                            resumptionStr = Integer.toString(resumptionInt);

                                            break;

                            case "discover":   // length of results unknown                                      

                                            if( (emptyXml.toLowerCase()).equals(resultXml.toLowerCase()) ) {

                                                resumptionStr = "";

                                                resultTotal   = resumptionInt;

                                            } else {

                                                resumptionStr = Integer.toString(resumptionInt);

                                                resultTotal   = resumptionInt+1;

                                            }

                                            break;

                            default:        // otherwise: abort

                                            resultTotal = resumptionInt;

                                            break;

                        }

			if (resultTotal == -1) {

				resultTotal = Integer.parseInt(xprResultTotalPath.evaluate(resultNode));

				log.info("resultTotal was -1 is now: " + resultTotal);

			}

			log.info("resultTotal: " + resultTotal);

			log.info("resInt: " + resumptionInt);

			if (resumptionInt < resultTotal) {

				nextQuery = baseUrl + "?" + queryParams + querySize + "&" + resumptionParam + "=" + resumptionStr + queryFormat;

			} else

				nextQuery = "";

                        log.debug("nextQueryUrl: " + nextQuery);

			return nextQuery;

		} catch(Exception e) {

			log.error(e);

			throw new IllegalStateException("collection failed: " + e.getMessage());

		}

	}

        /**

         * convert in Json-KeyName 'whitespace(s)' to '_' and '/' to '_', '(' and ')' to ''

         * 

         * @param jsonInput

         * @return 

         */

        private String syntaxConvertJsonKeyNamens(String jsonInput) {

            log.trace("before convertJsonKeyNames: " + jsonInput);

            // replace ' 's in JSON Namens with '_'

            while (jsonInput.matches(".*\"([^\"]*)\\s+([^\"]*)\":.*")) {

                jsonInput = jsonInput.replaceAll("\"([^\"]*)\\s+([^\"]*)\":", "\"$1_$2\":");

            }

            // replace forward-slash (sign '/' ) in JSON Names with '_'

            while (jsonInput.matches(".*\"([^\"]*)/([^\"]*)\":.*")) {

                jsonInput = jsonInput.replaceAll("\"([^\"]*)/([^\"]*)\":", "\"$1_$2\":");

            }

            // replace '(' in JSON Names with ''

            while (jsonInput.matches(".*\"([^\"]*)[(]([^\"]*)\":.*")) {

                jsonInput = jsonInput.replaceAll("\"([^\"]*)[(]([^\"]*)\":", "\"$1$2\":");

            }

            // replace ')' in JSON Names with ''

            while (jsonInput.matches(".*\"([^\"]*)[)]([^\"]*)\":.*")) {

                jsonInput = jsonInput.replaceAll("\"([^\"]*)[)]([^\"]*)\":", "\"$1$2\":");

            }

            log.trace("after syntaxConvertJsonKeyNames: " + jsonInput);

            return jsonInput;

        }

}