D-Net

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

import java.util.List;

import java.util.Map;

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 eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.msro.workflows.procs.Env;

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

import eu.dnetlib.rmi.data.hadoop.HadoopService;

import eu.dnetlib.rmi.enabling.ISLookUpException;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public abstract class PrepareIISParams extends SimpleJobNode {

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String clusterName;

	private String clusterParam = "cluster";

	private String zkQuorumParam = "export_action_hbase_remote_zookeeper_quorum";

	private String zkPortParam = "export_action_hbase_remote_zookeeper_clientport";

	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 = "";

	protected void prepare(final Env env) throws Exception {

		env.setAttribute(getClusterParam(), getClusterName());

		// Assumes we only have one mdStore service instance

		env.setAttribute(getMdStoreStoreLocationParam(), getServiceEndpoint(getXqueryMdStoreService()));

		// Assumes we only have one objectStore service instance

		env.setAttribute(getObjectStoreLocationParam(), getServiceEndpoint(getXqueryObjectStoreService()));

		env.setAttribute(getIslookupLocationParam(), getServiceEndpoint(getXqueryIsLookupService()));

		env.setAttribute(getImportProjectConceptsContextIdParam(), getImportProjectConceptsContextId());

		final Map<String, String> conf = serviceLocator.getService(HadoopService.class).getClusterConfiguration(getClusterName());

		final Map<String, String> exportConf = serviceLocator.getService(HadoopService.class).getClusterConfiguration(ClusterName.DM.toString());

		final String zkPort = exportConf.get("hbase.zookeeper.property.clientPort");

		final String zkQuorum = getZkQuorumCSV(exportConf, zkPort);

		env.setAttribute(getZkQuorumParam(), zkQuorum);

		env.setAttribute(getZkPortParam(), zkPort);

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

		env.setAttribute("nameNode", nameNode);

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

		env.setAttribute(getOozieWfAppPathParam(), getURI(nameNode, getOozieWfAppPath()));

		env.setAttribute(getMdStoreDatasetParam(), asCSV(getProfileIds(getXqueryDatasetStore())));

	}

	protected String getServiceEndpoint(final String xquery) throws MSROException {

		try {

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

		} catch (final ISLookUpException e) {

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

		} catch (final NoSuchElementException e) {

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

		} catch (final 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(this.serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery));

		} catch (final ISLookUpException e) {

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

		} catch (final NoSuchElementException e) {

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

		} catch (final 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 {

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

			if (ids.isEmpty()) {

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

			}

			return ids;

		} catch (final 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 Map<String, String> conf, final String zkPort) {

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

	}

	public String getXqueryMdStoreService() {

		return this.xqueryMdStoreService;

	}

	@Required

	public void setXqueryMdStoreService(final String xqueryMdStoreService) {

		this.xqueryMdStoreService = xqueryMdStoreService;

	}

	public String getMdStoreStoreLocationParam() {

		return this.mdStoreStoreLocationParam;

	}

	public void setMdStoreStoreLocationParam(final String mdStoreStoreLocationParam) {

		this.mdStoreStoreLocationParam = mdStoreStoreLocationParam;

	}

	public String getClusterName() {

		return this.clusterName;

	}

	public void setClusterName(final String clusterName) {

		this.clusterName = clusterName;

	}

	public String getZkQuorumParam() {

		return this.zkQuorumParam;

	}

	public void setZkQuorumParam(final String zkQuorumParam) {

		this.zkQuorumParam = zkQuorumParam;

	}

	public String getZkPortParam() {

		return this.zkPortParam;

	}

	public void setZkPortParam(final String zkPortParam) {

		this.zkPortParam = zkPortParam;

	}

	public String getClusterParam() {

		return this.clusterParam;

	}

	public void setClusterParam(final String clusterParam) {

		this.clusterParam = clusterParam;

	}

	public String getOozieWfAppPathParam() {

		return this.oozieWfAppPathParam;

	}

	public void setOozieWfAppPathParam(final String oozieWfAppPathParam) {

		this.oozieWfAppPathParam = oozieWfAppPathParam;

	}

	public String getOozieWfAppPath() {

		return this.oozieWfAppPath;

	}

	public void setOozieWfAppPath(final String oozieWfAppPath) {

		this.oozieWfAppPath = oozieWfAppPath;

	}

	@Required

	public String getXqueryDatasetStore() {

		return this.xqueryDatasetStore;

	}

	public void setXqueryDatasetStore(final String xqueryDatasetStore) {

		this.xqueryDatasetStore = xqueryDatasetStore;

	}

	public String getMdStoreDatasetParam() {

		return this.mdStoreDatasetParam;

	}

	public void setMdStoreDatasetParam(final String mdStoreDatasetParam) {

		this.mdStoreDatasetParam = mdStoreDatasetParam;

	}

	public String getXqueryObjectStoreService() {

		return this.xqueryObjectStoreService;

	}

	@Required

	public void setXqueryObjectStoreService(final String xqueryObjectStoreService) {

		this.xqueryObjectStoreService = xqueryObjectStoreService;

	}

	public String getObjectStoreLocationParam() {

		return this.objectStoreLocationParam;

	}

	public void setObjectStoreLocationParam(final String objectStoreLocationParam) {

		this.objectStoreLocationParam = objectStoreLocationParam;

	}

	public String getObjectStoreBlacklistCSV() {

		return this.objectStoreBlacklistCSV;

	}

	public void setObjectStoreBlacklistCSV(final String objectStoreBlacklistCSV) {

		this.objectStoreBlacklistCSV = objectStoreBlacklistCSV;

	}

	public String getXqueryIsLookupService() {

		return this.xqueryIsLookupService;

	}

	@Required

	public void setXqueryIsLookupService(final String xqueryIsLookupService) {

		this.xqueryIsLookupService = xqueryIsLookupService;

	}

	public String getIslookupLocationParam() {

		return this.islookupLocationParam;

	}

	public void setIslookupLocationParam(final String islookupLocationParam) {

		this.islookupLocationParam = islookupLocationParam;

	}

	public String getImportProjectConceptsContextIdParam() {

		return this.importProjectConceptsContextIdParam;

	}

	public void setImportProjectConceptsContextIdParam(final String importProjectConceptsContextIdParam) {

		this.importProjectConceptsContextIdParam = importProjectConceptsContextIdParam;

	}

	public String getImportProjectConceptsContextId() {

		return this.importProjectConceptsContextId;

	}

	public void setImportProjectConceptsContextId(final String importProjectConceptsContextId) {

		this.importProjectConceptsContextId = importProjectConceptsContextId;

	}

}