D-Net

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

import java.util.List;

import java.util.NoSuchElementException;

import javax.annotation.Resource;

import com.google.common.base.Joiner;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;

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

public abstract class PrepareIISParamsV2 extends SimpleJobNode {

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

	@Resource

	protected ConfigurationEnumerator configurationEnumerator;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String clusterName;

	private String clusterParam = "cluster";

	private String oozieWfAppPath;

	private String oozieWfAppPathParam = "oozie.wf.application.path";

	private String xqueryMdStoreService;

	private String mdStoreStoreLocationParam = "import_mdstore_service_location";

	private String xqueryObjectStoreService;

	private String objectStoreLocationParam = "import_content_object_store_location";

	private String xqueryIsLookupService;

	private String islookupLocationParam = "import_islookup_service_location";

	private String importProjectConceptsContextIdParam = "import_project_concepts_context_id";

	private String importProjectConceptsContextId;

	private String xqueryDatasetStore;

	private String mdStoreDatasetParam = "import_dataset_mdstore_ids_csv";

	private String objectStoreBlacklistCSV = "";

	private String importHbaseDumpLocationParam = "import_hbase_dump_location";

	private String importHbaseDumpLocation;

	protected void prepare(final NodeToken token) throws Exception {

		token.getEnv().setAttribute(getClusterParam(), getClusterName());

		// Assumes we only have one mdStore service instance

		token.getEnv().setAttribute(getMdStoreStoreLocationParam(), getServiceEndpoint(getXqueryMdStoreService()));

		// Assumes we only have one objectStore service instance

		token.getEnv().setAttribute(getObjectStoreLocationParam(), getServiceEndpoint(getXqueryObjectStoreService()));

		token.getEnv().setAttribute(getIslookupLocationParam(), getServiceEndpoint(getXqueryIsLookupService()));

		token.getEnv().setAttribute(getImportProjectConceptsContextIdParam(), getImportProjectConceptsContextId());

		Configuration dmConf = configurationEnumerator.get(ClusterName.DM);

		String dmNameNode = dmConf.get("fs.defaultFS");

		token.getEnv().setAttribute(getImportHbaseDumpLocationParam(), getURI(dmNameNode, getImportHbaseDumpLocation()));

		Configuration conf = configurationEnumerator.get(ClusterName.valueOf(getClusterName()));

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

		token.getEnv().setAttribute("nameNode", nameNode);

		token.getEnv().setAttribute("jobTracker", conf.get("mapred.job.tracker"));

		token.getEnv().setAttribute(getOozieWfAppPathParam(), getURI(nameNode, getOozieWfAppPath()));

		token.getEnv().setAttribute(getMdStoreDatasetParam(), asCSV(getProfileIds(getXqueryDatasetStore())));

	}

	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: " + getXqueryMdStoreService(), e);

		} catch (IllegalArgumentException e) {

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

		}

	}

	protected String getProfileId(final String xquery) throws MSROException {

		try {

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

		} catch (ISLookUpException e) {

			throw new MSROException("unable to fetch profile id", e);

		} catch (NoSuchElementException e) {

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

		} catch (IllegalArgumentException e) {

			throw new MSROException("more than one profile profiles was found, we assume to have only one available, xquery: " + xquery, e);

		}

	}

	protected List<String> getProfileIds(final String xquery) throws MSROException {

		try {

			List<String> ids = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

			if (ids.isEmpty()) {

				log.warn("couldn't find any profile, xquery: " + xquery);

			}

			return ids;

		} catch (ISLookUpException e) {

			throw new MSROException("unable to fetch profile ids, x query: " + xquery, e);

		}

	}

	protected String getFilteredObjectStoreCSV(final String xquery) throws MSROException {

		return asCSV(filter(getProfileIds(xquery), asList(getObjectStoreBlacklistCSV())));

	}

	protected List<String> filter(final List<String> list, final List<String> filter) {

		if (filter == null || filter.isEmpty()) { return list; }

		list.removeAll(filter);

		return list;

	}

	protected String asCSV(final List<String> list) {

		return Joiner.on(",").skipNulls().join(list);

	}

	protected List<String> asList(final String csv) {

		return Lists.newArrayList(Splitter.on(",").trimResults().omitEmptyStrings().split(csv));

	}

	private String getURI(final String nameNode, final String relative) {

		// TODO ensure to return a valid URI

		return nameNode + relative;

	}

	private String getZkQuorumCSV(final Configuration conf, final String zkPort) {

		return Joiner.on(":" + zkPort + ",").join(Splitter.on(",").omitEmptyStrings().split(conf.get("hbase.zookeeper.quorum")));

	}

	@Required

	public void setXqueryMdStoreService(final String xqueryMdStoreService) {

		this.xqueryMdStoreService = xqueryMdStoreService;

	}

	public String getXqueryMdStoreService() {

		return xqueryMdStoreService;

	}

	public String getMdStoreStoreLocationParam() {

		return mdStoreStoreLocationParam;

	}

	public void setMdStoreStoreLocationParam(final String mdStoreStoreLocationParam) {

		this.mdStoreStoreLocationParam = mdStoreStoreLocationParam;

	}

	public String getClusterName() {

		return clusterName;

	}

	public void setClusterName(final String clusterName) {

		this.clusterName = clusterName;

	}

	public String getClusterParam() {

		return clusterParam;

	}

	public void setClusterParam(final String clusterParam) {

		this.clusterParam = clusterParam;

	}

	public String getOozieWfAppPathParam() {

		return oozieWfAppPathParam;

	}

	public void setOozieWfAppPathParam(final String oozieWfAppPathParam) {

		this.oozieWfAppPathParam = oozieWfAppPathParam;

	}

	public String getOozieWfAppPath() {

		return oozieWfAppPath;

	}

	public void setOozieWfAppPath(final String oozieWfAppPath) {

		this.oozieWfAppPath = oozieWfAppPath;

	}

	@Required

	public String getXqueryDatasetStore() {

		return xqueryDatasetStore;

	}

	public void setXqueryDatasetStore(final String xqueryDatasetStore) {

		this.xqueryDatasetStore = xqueryDatasetStore;

	}

	public String getMdStoreDatasetParam() {

		return mdStoreDatasetParam;

	}

	public void setMdStoreDatasetParam(final String mdStoreDatasetParam) {

		this.mdStoreDatasetParam = mdStoreDatasetParam;

	}

	public String getXqueryObjectStoreService() {

		return xqueryObjectStoreService;

	}

	@Required

	public void setXqueryObjectStoreService(final String xqueryObjectStoreService) {

		this.xqueryObjectStoreService = xqueryObjectStoreService;

	}

	public String getObjectStoreLocationParam() {

		return objectStoreLocationParam;

	}

	public void setObjectStoreLocationParam(final String objectStoreLocationParam) {

		this.objectStoreLocationParam = objectStoreLocationParam;

	}

	public String getObjectStoreBlacklistCSV() {

		return objectStoreBlacklistCSV;

	}

	public void setObjectStoreBlacklistCSV(final String objectStoreBlacklistCSV) {

		this.objectStoreBlacklistCSV = objectStoreBlacklistCSV;

	}

	public String getXqueryIsLookupService() {

		return xqueryIsLookupService;

	}

	@Required

	public void setXqueryIsLookupService(final String xqueryIsLookupService) {

		this.xqueryIsLookupService = xqueryIsLookupService;

	}

	public String getIslookupLocationParam() {

		return islookupLocationParam;

	}

	public void setIslookupLocationParam(final String islookupLocationParam) {

		this.islookupLocationParam = islookupLocationParam;

	}

	public String getImportProjectConceptsContextIdParam() {

		return importProjectConceptsContextIdParam;

	}

	public void setImportProjectConceptsContextIdParam(final String importProjectConceptsContextIdParam) {

		this.importProjectConceptsContextIdParam = importProjectConceptsContextIdParam;

	}

	public String getImportProjectConceptsContextId() {

		return importProjectConceptsContextId;

	}

	public void setImportProjectConceptsContextId(final String importProjectConceptsContextId) {

		this.importProjectConceptsContextId = importProjectConceptsContextId;

	}

	public String getImportHbaseDumpLocationParam() {

		return importHbaseDumpLocationParam;

	}

	public void setImportHbaseDumpLocationParam(final String importHbaseDumpLocationParam) {

		this.importHbaseDumpLocationParam = importHbaseDumpLocationParam;

	}

	public String getImportHbaseDumpLocation() {

		return importHbaseDumpLocation;

	}

	public void setImportHbaseDumpLocation(final String importHbaseDumpLocation) {

		this.importHbaseDumpLocation = importHbaseDumpLocation;

	}

}