D-Net

package eu.dnetlib.msro.workflows.nodes.unpack;

import java.io.StringReader;

import java.util.Iterator;

import java.util.Queue;

import java.util.concurrent.PriorityBlockingQueue;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Required;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

public class UnpackJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	private String inputEprParam;

	private String outputEprParam;

	private String xpath;

	private IterableResultSetFactory iterableResultSetFactory;

	private ResultSetClientFactory resultSetClientFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final Iterator<String> client = resultSetClientFactory.getClient(token.getEnv().getAttribute(inputEprParam)).iterator();

		final Queue<String> queue = new PriorityBlockingQueue<String>();

		while (queue.isEmpty() && client.hasNext()) {

			populateQueue(queue, client.next(), xpath);

		}

		final W3CEndpointReference epr = iterableResultSetFactory.createIterableResultSet(new Iterable<String>() {

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						synchronized (queue) {

							return !queue.isEmpty();

						}

					}

					@Override

					public String next() {

						synchronized (queue) {

							final String res = queue.poll();

							while (queue.isEmpty() && client.hasNext()) {

								populateQueue(queue, client.next(), xpath);

							}

							return res;

						}

					}

					@Override

					public void remove() {}

				};

			}

		});

		token.getEnv().setAttribute(outputEprParam, epr.toString());

		return Arc.DEFAULT_ARC;

	}

	private void populateQueue(final Queue<String> queue, final String record, final String xpath) {

		try {

			final SAXReader reader = new SAXReader();

			final Document doc = reader.read(new StringReader(record));

			for (Object o : doc.selectNodes(xpath)) {

				queue.add(((Node) o).asXML());

			}

		} catch (Exception e) {

			log.error("Error unpacking record: \n" + record, e);

			throw new RuntimeException(e);

		}

	}

	public IterableResultSetFactory getIterableResultSetFactory() {

		return iterableResultSetFactory;

	}

	@Required

	public void setIterableResultSetFactory(final IterableResultSetFactory iterableResultSetFactory) {

		this.iterableResultSetFactory = iterableResultSetFactory;

	}

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	@Required

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getXpath() {

		return xpath;

	}

	public void setXpath(final String xpath) {

		this.xpath = xpath;

	}

}