D-Net

package eu.dnetlib.msro.workflows.hadoop;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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.rmi.MSROException;

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

public class PrepareMDStoreImportJobNode extends SimpleJobNode {

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

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private String hdfsPathParam;

	private String hdfsPath;

	private String mappingParam;

	private String mapping;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		token.getEnv().setAttribute(getHdfsPathParam(), getHdfsPath());

		token.getEnv().setAttribute(getMappingParam(), readXslt(getMapping()));

		return Arc.DEFAULT_ARC;

	}

	private String readXslt(final String profileId) throws IOException, MSROException, ISLookUpDocumentNotFoundException, ISLookUpException {

		if (StringUtils.isBlank(profileId)) throw new MSROException("missing profile id");

		log.info("loading mapping from profile id: " + profileId);

		final String xquery =

				String.format("/RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='%s']/BODY/CONFIGURATION/SCRIPT/CODE/*[local-name()='stylesheet']", profileId);

		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);

	}

	public String getHdfsPathParam() {

		return hdfsPathParam;

	}

	public void setHdfsPathParam(final String hdfsPathParam) {

		this.hdfsPathParam = hdfsPathParam;

	}

	public String getHdfsPath() {

		return hdfsPath;

	}

	public void setHdfsPath(final String hdfsPath) {

		this.hdfsPath = hdfsPath;

	}

	public String getMapping() {

		return mapping;

	}

	public void setMapping(final String mapping) {

		this.mapping = mapping;

	}

	public String getMappingParam() {

		return mappingParam;

	}

	public void setMappingParam(final String mappingParam) {

		this.mappingParam = mappingParam;

	}

}