D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.datacite;

import java.io.StringReader;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

import javax.annotation.Resource;

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.io.SAXReader;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.collector.plugins.datasets.DatasetsByJournalIterator;

import eu.dnetlib.data.collector.plugins.datasets.PangaeaJournalInfo;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

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

import eu.dnetlib.enabling.resultset.client.utils.EPRUtils;

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

// TODO: Auto-generated Javadoc

/**

 * The Class CollectDatasetsByJournalJobNode.

 */

public class CollectDatasetsByJournalJobNode extends SimpleJobNode {

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

	public static PangaeaJournalInfo END_QUEUE = new PangaeaJournalInfo();

	/** The datasource id. */

	private String datasourceId;

	/** The output epr param. */

	private String outputEprParam;

	/** The input epr param. */

	private String inputEprParam;

	/** The result set client factory. */

	private ResultSetClientFactory resultSetClientFactory;

	/** The result set factory. */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	/** The executor. */

	private Executor executor = Executors.newSingleThreadExecutor();

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.nodes.SimpleJobNode#execute(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final W3CEndpointReference inputEpr = new EPRUtils().getEpr(token.getEnv().getAttribute(inputEprParam));

		final Iterable<String> input = resultSetClientFactory.getClient(inputEpr);

		final BlockingQueue<PangaeaJournalInfo> publicationsQueue = new ArrayBlockingQueue<PangaeaJournalInfo>(50);

		executor.execute(new Runnable() {

			@Override

			public void run() {

				final SAXReader reader = new SAXReader();

				for (String inputString : input) {

					try {

						Document doc = reader.read(new StringReader(inputString));

						final String jId = doc.valueOf("//FIELD[@name='id']");

						final String dsId = doc.valueOf("//FIELD[@name='datasource']");

						final String dsName = doc.valueOf("//FIELD[@name='name']");

						final String jISSN = doc.valueOf("//FIELD[@name='issn']");

						PangaeaJournalInfo info = new PangaeaJournalInfo();

						info.setDatasourceId(dsId);

						info.setJournalId(jId);

						info.setJournalName(dsName);

						info.setJournalISSN(jISSN);

						publicationsQueue.put(info);

					} catch (Exception e) {

						log.error(e);

					}

				}

				try {

					publicationsQueue.put(END_QUEUE);

				} catch (InterruptedException e) {

					log.error(e);

				}

			}

		});

		IteratorOnQueue<PangaeaJournalInfo> itOnQueue = new IteratorOnQueue<PangaeaJournalInfo>(publicationsQueue, CollectDatasetsByJournalJobNode.END_QUEUE);

		W3CEndpointReference eprOutput = resultSetFactory.createIterableResultSet(new DatasetsByJournalIterator(itOnQueue));

		token.getEnv().setAttribute(getOutputEprParam(), eprOutput.toString());

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Gets the datasource id.

	 *

	 * @return the datasourceId

	 */

	public String getDatasourceId() {

		return datasourceId;

	}

	/**

	 * Sets the datasource id.

	 *

	 * @param datasourceId

	 *            the datasourceId to set

	 */

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	/**

	 * @return the outputEprParam

	 */

	public String getOutputEprParam() {

		return outputEprParam;

	}

	/**

	 * @param outputEprParam

	 *            the outputEprParam to set

	 */

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	/**

	 * @return the inputEprParam

	 */

	public String getInputEprParam() {

		return inputEprParam;

	}

	/**

	 * @param inputEprParam

	 *            the inputEprParam to set

	 */

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	/**

	 * Gets the result set client factory.

	 *

	 * @return the resultSetClientFactory

	 */

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	/**

	 * Sets the result set client factory.

	 *

	 * @param resultSetClientFactory

	 *            the resultSetClientFactory to set

	 */

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

}