D-Net

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

import java.util.NoSuchElementException;

import javax.annotation.Resource;

import com.google.common.collect.Iterables;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;

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.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.hadoop.SubmitHadoopJobNode;

import org.apache.hadoop.conf.Configuration;

/**

 * Created by claudio on 07/09/16.

 */

public class IISCacheBuilderJobNode extends SubmitHadoopJobNode {

	@Resource

	protected ConfigurationEnumerator configurationEnumerator;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String xqueryObjectStoreService;

	private String oozieWfAppPath;

	private String objectStoreId;

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		final Configuration conf = configurationEnumerator.get(ClusterName.valueOf(getCluster()));

		String nameNode = conf.get("fs.defaultFS");

		job.getParameters().put("nameNode", nameNode);

		job.getParameters().put("jobTracker", conf.get("mapred.job.tracker"));

		job.getParameters().put("objectstore_service_location", getServiceEndpoint(getXqueryObjectStoreService()));

		job.getParameters().put("approved_objectstores_csv", getObjectStoreId());

		super.prepareJob(job, token);

	}

	protected String getServiceEndpoint(final String xquery) throws MSROException {

		try {

			return Iterables.getOnlyElement(serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery));

		} catch (ISLookUpException e) {

			throw new MSROException("unable to fetch service endpoint", e);

		} catch (NoSuchElementException e) {

			throw new MSROException("unable to find service endpoint, xquery: " + xquery, e);

		} catch (IllegalArgumentException e) {

			throw new MSROException("more than one services found, we assume to have only one available", e);

		}

	}

	public String getOozieWfAppPath() {

		return oozieWfAppPath;

	}

	public void setOozieWfAppPath(final String oozieWfAppPath) {

		this.oozieWfAppPath = oozieWfAppPath;

	}

	public String getObjectStoreId() {

		return objectStoreId;

	}

	public void setObjectStoreId(final String objectStoreId) {

		this.objectStoreId = objectStoreId;

	}

	public String getXqueryObjectStoreService() {

		return xqueryObjectStoreService;

	}

	public void setXqueryObjectStoreService(final String xqueryObjectStoreService) {

		this.xqueryObjectStoreService = xqueryObjectStoreService;

	}

}