D-Net

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

import java.io.StringReader;

import java.util.Map;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

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

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

import com.google.common.collect.Maps;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.database.rmi.DatabaseException;

import eu.dnetlib.enabling.database.rmi.DatabaseService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.resultset.MappedResultSetFactory;

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

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

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

public class PatchHostedByJobNode extends SimpleJobNode {

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

	private String inputEprParam;

	private String outputEprParam;

	@Value("${dnet.openaire.db.name}")

	private String dbName;

	private String countersParam;

	private String hostedbyMapTable;

	private String xpathEntry;

	private String overrideDataSourceId;

	private String keyTypeFilter;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private MappedResultSetFactory mappedResultSetFactory;

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * {@inheritDoc}

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

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

		final HostedByCounters counters = new HostedByCounters();

		String datasourceId;

		if (StringUtils.isEmpty(overrideDataSourceId)) {

			datasourceId = token.getEnv().getAttribute("parentDatasourceId");

		} else {

			datasourceId = getOverrideDataSourceId();

		}

		Map<String, HostedByEntry> hostedByEntryMap;

		if (!StringUtils.isBlank(keyTypeFilter)) {

			hostedByEntryMap = loadHostedByMapFromType(keyTypeFilter);

		} else {

			hostedByEntryMap = loadHostedByMap(datasourceId);

		}

		final W3CEndpointReference epr = mappedResultSetFactory.createMappedResultSet(inputEpr, new PatchHostedBy(hostedByEntryMap,

				getXpathEntry(), counters));

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

		token.getEnv().setTransientAttribute(countersParam, counters);

		return Arc.DEFAULT_ARC;

	}

	private Map<String, HostedByEntry> loadHostedByMap(final String datasourceId) throws DocumentException, DatabaseException {

		final String sql = "SELECT d.id, d.officialname, p.entry from %s p JOIN dsm_datasources d ON (p.datasourceid = d.id) WHERE p.oa_source_id= '%s'";

		return getHostedByEntryMapByQuery(datasourceId, sql);

	}

	private Map<String, HostedByEntry> loadHostedByMapFromType(final String keyType) throws DocumentException, DatabaseException {

		final String sql = "SELECT d.id, d.officialname, p.entry from %s p JOIN dsm_datasources d ON (p.datasourceid = d.id) WHERE p.key_type= '%s'";

		return getHostedByEntryMapByQuery(keyType, sql);

	}

	private Map<String, HostedByEntry> getHostedByEntryMapByQuery(final String datasourceId, final String sql)

			throws DatabaseException, DocumentException {

		final Map<String, HostedByEntry> map = Maps.newHashMap();

		final W3CEndpointReference epr = serviceLocator.getService(DatabaseService.class).searchSQL(getDbName(),

				String.format(sql, getHostedbyMapTable(), datasourceId));

		final SAXReader reader = new SAXReader();

		for (final String s : resultSetClientFactory.getClient(epr)) {

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

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

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

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

			map.put(entry, new HostedByEntry(dsId, dsName));

		}

		log.info(String.format("built hostedByMap from dsId '%s', size: '%s'", datasourceId, map.size()));

		return map;

	}

	/**

	 * Getter for property 'inputEprParam'.

	 *

	 * @return Value for property 'inputEprParam'.

	 */

	public String getInputEprParam() {

		return inputEprParam;

	}

	/**

	 * Setter for property 'inputEprParam'.

	 *

	 * @param inputEprParam

	 *            Value to set for property 'inputEprParam'.

	 */

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	/**

	 * Getter for property 'outputEprParam'.

	 *

	 * @return Value for property 'outputEprParam'.

	 */

	public String getOutputEprParam() {

		return outputEprParam;

	}

	/**

	 * Setter for property 'outputEprParam'.

	 *

	 * @param outputEprParam

	 *            Value to set for property 'outputEprParam'.

	 */

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	/**

	 * Getter for property 'dbName'.

	 *

	 * @return Value for property 'dbName'.

	 */

	public String getDbName() {

		return dbName;

	}

	/**

	 * Setter for property 'dbName'.

	 *

	 * @param dbName

	 *            Value to set for property 'dbName'.

	 */

	public void setDbName(final String dbName) {

		this.dbName = dbName;

	}

	public String getKeyTypeFilter() {

		return keyTypeFilter;

	}

	public void setKeyTypeFilter(final String keyTypeFilter) {

		this.keyTypeFilter = keyTypeFilter;

	}

	/**

	 * Getter for property 'mappedResultSetFactory'.

	 *

	 * @return Value for property 'mappedResultSetFactory'.

	 */

	public MappedResultSetFactory getMappedResultSetFactory() {

		return mappedResultSetFactory;

	}

	/**

	 * Setter for property 'mappedResultSetFactory'.

	 *

	 * @param mappedResultSetFactory

	 *            Value to set for property 'mappedResultSetFactory'.

	 */

	@Required

	public void setMappedResultSetFactory(final MappedResultSetFactory mappedResultSetFactory) {

		this.mappedResultSetFactory = mappedResultSetFactory;

	}

	/**

	 * Getter for property 'resultSetClientFactory'.

	 *

	 * @return Value for property 'resultSetClientFactory'.

	 */

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	/**

	 * Setter for property 'resultSetClientFactory'.

	 *

	 * @param resultSetClientFactory

	 *            Value to set for property 'resultSetClientFactory'.

	 */

	@Required

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	/**

	 * Getter for property 'countersParam'.

	 *

	 * @return Value for property 'countersParam'.

	 */

	public String getCountersParam() {

		return countersParam;

	}

	/**

	 * Setter for property 'countersParam'.

	 *

	 * @param countersParam

	 *            Value to set for property 'countersParam'.

	 */

	public void setCountersParam(final String countersParam) {

		this.countersParam = countersParam;

	}

	/**

	 * @return the hostedbyMapTable

	 */

	public String getHostedbyMapTable() {

		return hostedbyMapTable;

	}

	/**

	 * @param hostedbyMapTable

	 *            the hostedbyMapTable to set

	 */

	public void setHostedbyMapTable(final String hostedbyMapTable) {

		this.hostedbyMapTable = hostedbyMapTable;

	}

	public String getXpathEntry() {

		return xpathEntry;

	}

	public void setXpathEntry(final String xpathEntry) {

		this.xpathEntry = xpathEntry;

	}

	/**

	 * @return the overrideDataSourceId

	 */

	public String getOverrideDataSourceId() {

		return overrideDataSourceId;

	}

	/**

	 * @param overrideDataSourceId

	 *            the overrideDataSourceId to set

	 */

	public void setOverrideDataSourceId(final String overrideDataSourceId) {

		this.overrideDataSourceId = overrideDataSourceId;

	}

}