D-Net

package eu.dnetlib.msro.workflows.hadoop;

import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

/**

 * This jobNode resolves the original datasource ids to the list of md store ids holding the relative cleaned records.

 *

 * @author claudio

 *

 */

public class MDStoreDatasourceResolverJobNode extends SimpleJobNode {

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

	private String originalDatasourceIdsCSV;

	@Value("${dnet.openaire.dataload.datasource.mdstores.xquery}")

	private String xqueryTemplate;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	public String execute(final NodeToken token) throws Exception {

		log.info("resolving MDStore ids for datasources: " + getOriginalDatasourceIdsCSV());

		final List<String> mdIds = Lists.newArrayList();

		for (final String originalId : splitter().split(getOriginalDatasourceIdsCSV())) {

			mdIds.addAll(resolveMdIds(originalId));

		}

		log.info(String.format("adding %s mdStore ids in wf env", mdIds.size()));

		token.getEnv().setAttribute("mdId", new Gson().toJson(mdIds));

		return Arc.DEFAULT_ARC;

	}

	// @Cacheable(value = "mdIds", key = "#acronym")

	public List<String> resolveMdIds(final String id) throws ISLookUpException {

		log.info("Resolving mdID for " + id + ". Cache not used.");

		final String xQuery = String.format(xqueryTemplate, id);

		final List<String> mdIds = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xQuery);

		return mdIds;

	}

	private Splitter splitter() {

		return Splitter.on(",").trimResults().omitEmptyStrings();

	}

	public String getOriginalDatasourceIdsCSV() {

		return originalDatasourceIdsCSV;

	}

	public void setOriginalDatasourceIdsCSV(final String originalDatasourceIdsCSV) {

		this.originalDatasourceIdsCSV = originalDatasourceIdsCSV;

	}

}